From dbc65707c1acab3e7c533468d3f085434fa79b39 Mon Sep 17 00:00:00 2001 From: Jordvl Date: Fri, 17 Jul 2026 17:32:14 +0200 Subject: [PATCH 1/8] Add ND1000 open science project backend support --- .../ThermoFisherNanoDrop1000_PLR_V1.py | 244 ++++++++++++++++++ pylabrobot/thermo_fisher/bosdescriptor.reg | 3 + .../thermo_fisher/nanodrop_setup_guide.md | 88 +++++++ 3 files changed, 335 insertions(+) create mode 100644 pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py create mode 100644 pylabrobot/thermo_fisher/bosdescriptor.reg create mode 100644 pylabrobot/thermo_fisher/nanodrop_setup_guide.md diff --git a/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py new file mode 100644 index 00000000000..1d127f218d7 --- /dev/null +++ b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py @@ -0,0 +1,244 @@ +import usb.core +import usb.util +import asyncio +import numpy as np +from typing import Tuple, List +# PyLabRobot core imports (Assuming standard PLR v1b1 structure) +from pylabrobot.device import Device, Driver +from pylabrobot.capabilities import CapabilityBackend +# Assuming an Absorbance Capability exists in PLR; if not, you would define it in capabilities/ +# from pylabrobot.capabilities.absorbance import AbsorbanceBackend, Absorbance + +class ThermoFisherNanoDrop1000Driver(Driver): + """ + Pure transport layer for the NanoDrop 1000. + Handles USB connections, endpoints, and raw byte transfers. + """ + VID = 0x2457 + PID = 0x1002 + EP_OUT = 0x02 + EP_IN_HEAVY = 0x82 + EP_IN_COMM = 0x87 + + def __init__(self): + super().__init__() + self.dev = None + + async def setup(self): + """Initializes the USB connection.""" + print("Connecting to NanoDrop...") + self.dev = usb.core.find(idVendor=self.VID, idProduct=self.PID) + if self.dev is None: + raise RuntimeError("NanoDrop not found. Is Zadig set to libusb-win32?") + + self.dev.set_configuration() + self.dev.clear_halt(self.EP_OUT) + self.dev.clear_halt(self.EP_IN_HEAVY) + self.dev.clear_halt(self.EP_IN_COMM) + + # Wake & Init + await self.send_command([0x08]) + await asyncio.sleep(0.1) + await self.send_command([0x01]) + await asyncio.sleep(0.2) + + async def stop(self): + """Safely powers down hardware and releases the USB port.""" + if self.dev: + try: + # Ensure lamp and magnet are off before disconnect + await self.send_command([0x03, 0x00]) + await self.send_command([0x0F, 0x00]) + self.dev.reset() + usb.util.dispose_resources(self.dev) + except Exception: + pass + print("NanoDrop safely disconnected.") + + async def send_command(self, payload: List[int]): + """Generic transport method for writing to the command mailbox.""" + self.dev.write(self.EP_OUT, payload) + + async def read_comm(self, timeout=500) -> bytes: + """Reads from the 64-byte text/status endpoint.""" + return self.dev.read(self.EP_IN_COMM, 64, timeout=timeout) + + 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): + data_buffer.extend(self.dev.read(self.EP_IN_HEAVY, 64, timeout=timeout)) + return data_buffer + + def flush_comm(self): + try: + while True: self.dev.read(self.EP_IN_COMM, 64, timeout=50) + except usb.core.USBTimeoutError: pass + + def flush_heavy(self): + try: + while True: self.dev.read(self.EP_IN_HEAVY, 512, timeout=50) + except usb.core.USBTimeoutError: pass + + +class ThermoFisherNanoDrop1000AbsorbanceBackend(CapabilityBackend): # Ideally inherits from AbsorbanceBackend + """ + Translates scientific workflow methods into raw driver commands. + Holds state for coefficients, dark spectra, and blank spectra. + """ + def __init__(self, driver: ThermoFisherNanoDrop1000Driver): + super().__init__() + self.driver = driver + self.coefficients = {} + self.wavelengths = None + + self.dark_spectrum = None + self.blank_spectrum = None + + async def _on_setup(self): + """Lifecycle hook to download factory calibration on boot.""" + await self._download_all_coefficients() + self._calculate_x_axis() + print("NanoDrop Initialized and Calibrated.") + + async def _on_stop(self): + """Lifecycle hook to clean up state on teardown.""" + self.coefficients.clear() + + async def set_lamp(self, state: bool): + cmd = 0xFF if state else 0x00 + await self.driver.send_command([0x03, cmd]) + + async def set_magnet(self, state: bool): + cmd = 0xFF if state else 0x00 + await self.driver.send_command([0x0F, cmd]) + + async def set_integration_time(self, ms: int): + if ms < 3: + ms = 3 + print('Integration too low, setting to 3 ms') + elif ms > 65535: + ms = 65535 + print('Integration too high, setting to 65535 ms') + + lsb = ms & 0xFF + msb = (ms >> 8) & 0xFF + await self.driver.send_command([0x02, lsb, msb]) + + async def _download_all_coefficients(self): + print("Downloading Factory Memory Map...") + self.driver.flush_comm() + + for index in range(1, 15): + if index == 5: continue + await self.driver.send_command([0x05, index]) + await asyncio.sleep(0.05) + try: + data = await self.driver.read_comm() + text = bytearray(data[2:]).decode('ascii', errors='ignore').split('\x00')[0] + self.coefficients[index] = float(text) + except Exception: + print(f"Warning: Failed to read coefficient index {index}") + + def _calculate_x_axis(self): + pixels = np.arange(2048) + c0, c1 = self.coefficients.get(1, 0), self.coefficients.get(2, 0) + c2, c3 = self.coefficients.get(3, 0), self.coefficients.get(4, 0) + self.wavelengths = c0 + (c1 * pixels) + (c2 * (pixels**2)) + (c3 * (pixels**3)) + + async def get_raw_spectrum(self) -> np.ndarray: + self.driver.flush_heavy() + await self.driver.send_command([0x09]) + + data_buffer = await self.driver.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 = np.array(pixels, dtype=float) + + # 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) + print("Acquiring Dark baseline...") + self.dark_spectrum = await self.get_raw_spectrum() + + await self.set_lamp(True) + await asyncio.sleep(0.2) + print("Acquiring Blank baseline...") + self.blank_spectrum = await self.get_raw_spectrum() + + await self.set_lamp(False) + await self.set_magnet(False) + print("Blanking complete.") + + async def measure_absorbance(self, integration_ms=20) -> Tuple[np.ndarray, np.ndarray]: + 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) + + print("Measuring sample...") + sample_spectrum = await self.get_raw_spectrum() + + await self.set_lamp(False) + await self.set_magnet(False) + + numerator = np.clip(sample_spectrum - self.dark_spectrum, 1, None) + denominator = np.clip(self.blank_spectrum - self.dark_spectrum, 1, None) + + transmittance = numerator / denominator + absorbance = -np.log10(transmittance) + + return self.wavelengths, absorbance + + +class ThermoFisherNanoDrop1000(Device): + """ + Main PyLabRobot Device Class. + Constructs the driver and registers the absorbance capability. + """ + def __init__(self, name: str = "NanoDrop1000"): + super().__init__(name=name) + + # Construct ONE driver + self.driver = ThermoFisherNanoDrop1000Driver() + + # Construct backends sharing the single driver + self.absorbance_backend = ThermoFisherNanoDrop1000AbsorbanceBackend(driver=self.driver) + + # Append to capabilities + self._capabilities.append(self.absorbance_backend) + + # setup() and stop() are completely removed! Inherited behavior takes over. + + # CAUTION: Convenience methods below map to Capability operations. + async def take_blank(self, integration_ms=20): + await self.absorbance_backend.take_blank(integration_ms) + + async def measure_absorbance(self, integration_ms=20): + return await self.absorbance_backend.measure_absorbance(integration_ms) \ No newline at end of file diff --git a/pylabrobot/thermo_fisher/bosdescriptor.reg b/pylabrobot/thermo_fisher/bosdescriptor.reg new file mode 100644 index 00000000000..ab4d5c453ba --- /dev/null +++ b/pylabrobot/thermo_fisher/bosdescriptor.reg @@ -0,0 +1,3 @@ +Windows Registry Editor Version 5.00 +[HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\usbflags\245710020002] +"SkipBOSDescriptorQuery"=dword:00000001 \ No newline at end of file diff --git a/pylabrobot/thermo_fisher/nanodrop_setup_guide.md b/pylabrobot/thermo_fisher/nanodrop_setup_guide.md new file mode 100644 index 00000000000..8f40bc74b3a --- /dev/null +++ b/pylabrobot/thermo_fisher/nanodrop_setup_guide.md @@ -0,0 +1,88 @@ +``` +The NanoDrop Open-Source Installation Guide +``` + +``` +Step 1: The BOS Descriptor Fix (Windows 10/11) Because the NanoDrop uses an +older USB 1.1 microcontroller, plugging it into a modern Windows 10/11 system +can cause Windows to request a "BOS Descriptor"—a feature that didn't exist when +the machine was built. This causes Windows to flag it as an "Unknown Device." +1.Download the SkipBOSDescriptor.reg file from the GitHub repository. +``` + +`2. Double-click the .reg file and click Yes on the Administrator prompt to merge it into your registry.` + +``` +Important Note: If Windows opens the file in Notepad instead of running +it, Windows is likely hiding file extensions and saved it as a .txt file. To +fix this: Open Windows File Explorer, click the View tab at the top, and check +the box for File name extensions. Rename the downloaded file to ensure it ends +in .reg (not .reg.txt). +``` + +``` +Further, sometimes windows will not let you open .reg files regularly, in +this case you could try: +``` + + `1. Press Win + R to open the Run dialogue box.` + + `2. Type 'regedit' and press Enter to open the Registry Editor.` + + `3. In the top menu, click File > Import.` + + `4. Locate your .reg file, select it, and click Open.` + +`3. Unplug the NanoDrop and plug it back in. It will now be recognized.` + +``` +Step 2: Installing the Python Driver (Zadig) To control the machine with Python, +we must temporarily replace the official driver with an open-source one. +``` + +`1. Download and run Zadig (zadig.akeo.ie).` + +`2. Go to Options -> List All Devices.` + +`3. Select the NanoDrop 1000 from the main dropdown menu.` + +``` +4.On the right side of the green arrow, use the up/down arrows to select +libusb-win32. +``` + +``` +5.Click Replace Driver (or Install Driver) and wait for the "Success" +message. +``` + +`6. Your Python script can now communicate with the hardware.` + +``` +Reverting to the Official NanoDrop Software +``` + +``` +If you wish to revert to the original software, you can seamlessly swap back to +the proprietary driver without uninstalling anything. +``` + +`1. Open Windows Device Manager.` + +`2. Scroll down and find the NanoDrop 1000 (it will likely be under "libusbwin32 devices").` + +`3. Right-click the device and select Update driver.` + +`4. Click Browse my computer for drivers.` + +`5. Click Let me pick from a list of available drivers on my computer.` + +`6. You will see a list containing the driver you just installed (libusbwin32) and the original official driver (often named something like NanoDrop 1000 Spectrometer or Cypress EZ-USB).` + +`7. Select the original official driver and click Next.` + +``` +8.Wait a few seconds for Windows to swap them over. You can now open the +official NanoDrop software. +``` + From adf87d904a823f46a8b3661df6ff689124a719a2 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Sat, 15 Aug 2026 20:41:06 -0700 Subject: [PATCH 2/8] Format NanoDrop backend --- .../ThermoFisherNanoDrop1000_PLR_V1.py | 483 +++++++++--------- 1 file changed, 249 insertions(+), 234 deletions(-) diff --git a/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py index 1d127f218d7..30d8ac13be5 100644 --- a/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py +++ b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py @@ -1,244 +1,259 @@ -import usb.core -import usb.util import asyncio +from typing import List, Tuple + import numpy as np -from typing import Tuple, List +import usb.core +import usb.util + +from pylabrobot.capabilities import CapabilityBackend + # PyLabRobot core imports (Assuming standard PLR v1b1 structure) from pylabrobot.device import Device, Driver -from pylabrobot.capabilities import CapabilityBackend + # Assuming an Absorbance Capability exists in PLR; if not, you would define it in capabilities/ -# from pylabrobot.capabilities.absorbance import AbsorbanceBackend, Absorbance +# from pylabrobot.capabilities.absorbance import AbsorbanceBackend, Absorbance + class ThermoFisherNanoDrop1000Driver(Driver): - """ - Pure transport layer for the NanoDrop 1000. - Handles USB connections, endpoints, and raw byte transfers. - """ - VID = 0x2457 - PID = 0x1002 - EP_OUT = 0x02 - EP_IN_HEAVY = 0x82 - EP_IN_COMM = 0x87 - - def __init__(self): - super().__init__() - self.dev = None - - async def setup(self): - """Initializes the USB connection.""" - print("Connecting to NanoDrop...") - self.dev = usb.core.find(idVendor=self.VID, idProduct=self.PID) - if self.dev is None: - raise RuntimeError("NanoDrop not found. Is Zadig set to libusb-win32?") - - self.dev.set_configuration() - self.dev.clear_halt(self.EP_OUT) - self.dev.clear_halt(self.EP_IN_HEAVY) - self.dev.clear_halt(self.EP_IN_COMM) - - # Wake & Init - await self.send_command([0x08]) - await asyncio.sleep(0.1) - await self.send_command([0x01]) - await asyncio.sleep(0.2) - - async def stop(self): - """Safely powers down hardware and releases the USB port.""" - if self.dev: - try: - # Ensure lamp and magnet are off before disconnect - await self.send_command([0x03, 0x00]) - await self.send_command([0x0F, 0x00]) - self.dev.reset() - usb.util.dispose_resources(self.dev) - except Exception: - pass - print("NanoDrop safely disconnected.") - - async def send_command(self, payload: List[int]): - """Generic transport method for writing to the command mailbox.""" - self.dev.write(self.EP_OUT, payload) - - async def read_comm(self, timeout=500) -> bytes: - """Reads from the 64-byte text/status endpoint.""" - return self.dev.read(self.EP_IN_COMM, 64, timeout=timeout) - - 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): - data_buffer.extend(self.dev.read(self.EP_IN_HEAVY, 64, timeout=timeout)) - return data_buffer - - def flush_comm(self): - try: - while True: self.dev.read(self.EP_IN_COMM, 64, timeout=50) - except usb.core.USBTimeoutError: pass - - def flush_heavy(self): - try: - while True: self.dev.read(self.EP_IN_HEAVY, 512, timeout=50) - except usb.core.USBTimeoutError: pass - - -class ThermoFisherNanoDrop1000AbsorbanceBackend(CapabilityBackend): # Ideally inherits from AbsorbanceBackend - """ - Translates scientific workflow methods into raw driver commands. - Holds state for coefficients, dark spectra, and blank spectra. - """ - def __init__(self, driver: ThermoFisherNanoDrop1000Driver): - super().__init__() - self.driver = driver - self.coefficients = {} - self.wavelengths = None - - self.dark_spectrum = None - self.blank_spectrum = None - - async def _on_setup(self): - """Lifecycle hook to download factory calibration on boot.""" - await self._download_all_coefficients() - self._calculate_x_axis() - print("NanoDrop Initialized and Calibrated.") - - async def _on_stop(self): - """Lifecycle hook to clean up state on teardown.""" - self.coefficients.clear() - - async def set_lamp(self, state: bool): - cmd = 0xFF if state else 0x00 - await self.driver.send_command([0x03, cmd]) - - async def set_magnet(self, state: bool): - cmd = 0xFF if state else 0x00 - await self.driver.send_command([0x0F, cmd]) - - async def set_integration_time(self, ms: int): - if ms < 3: - ms = 3 - print('Integration too low, setting to 3 ms') - elif ms > 65535: - ms = 65535 - print('Integration too high, setting to 65535 ms') - - lsb = ms & 0xFF - msb = (ms >> 8) & 0xFF - await self.driver.send_command([0x02, lsb, msb]) - - async def _download_all_coefficients(self): - print("Downloading Factory Memory Map...") - self.driver.flush_comm() - - for index in range(1, 15): - if index == 5: continue - await self.driver.send_command([0x05, index]) - await asyncio.sleep(0.05) - try: - data = await self.driver.read_comm() - text = bytearray(data[2:]).decode('ascii', errors='ignore').split('\x00')[0] - self.coefficients[index] = float(text) - except Exception: - print(f"Warning: Failed to read coefficient index {index}") - - def _calculate_x_axis(self): - pixels = np.arange(2048) - c0, c1 = self.coefficients.get(1, 0), self.coefficients.get(2, 0) - c2, c3 = self.coefficients.get(3, 0), self.coefficients.get(4, 0) - self.wavelengths = c0 + (c1 * pixels) + (c2 * (pixels**2)) + (c3 * (pixels**3)) - - async def get_raw_spectrum(self) -> np.ndarray: - self.driver.flush_heavy() - await self.driver.send_command([0x09]) - - data_buffer = await self.driver.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 = np.array(pixels, dtype=float) - - # 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) - print("Acquiring Dark baseline...") - self.dark_spectrum = await self.get_raw_spectrum() - - await self.set_lamp(True) - await asyncio.sleep(0.2) - print("Acquiring Blank baseline...") - self.blank_spectrum = await self.get_raw_spectrum() - - await self.set_lamp(False) - await self.set_magnet(False) - print("Blanking complete.") - - async def measure_absorbance(self, integration_ms=20) -> Tuple[np.ndarray, np.ndarray]: - 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) - - print("Measuring sample...") - sample_spectrum = await self.get_raw_spectrum() - - await self.set_lamp(False) - await self.set_magnet(False) - - numerator = np.clip(sample_spectrum - self.dark_spectrum, 1, None) - denominator = np.clip(self.blank_spectrum - self.dark_spectrum, 1, None) - - transmittance = numerator / denominator - absorbance = -np.log10(transmittance) - - return self.wavelengths, absorbance + """ + Pure transport layer for the NanoDrop 1000. + Handles USB connections, endpoints, and raw byte transfers. + """ + + VID = 0x2457 + PID = 0x1002 + EP_OUT = 0x02 + EP_IN_HEAVY = 0x82 + EP_IN_COMM = 0x87 + + def __init__(self): + super().__init__() + self.dev = None + + async def setup(self): + """Initializes the USB connection.""" + print("Connecting to NanoDrop...") + self.dev = usb.core.find(idVendor=self.VID, idProduct=self.PID) + if self.dev is None: + raise RuntimeError("NanoDrop not found. Is Zadig set to libusb-win32?") + + self.dev.set_configuration() + self.dev.clear_halt(self.EP_OUT) + self.dev.clear_halt(self.EP_IN_HEAVY) + self.dev.clear_halt(self.EP_IN_COMM) + + # Wake & Init + await self.send_command([0x08]) + await asyncio.sleep(0.1) + await self.send_command([0x01]) + await asyncio.sleep(0.2) + + async def stop(self): + """Safely powers down hardware and releases the USB port.""" + if self.dev: + try: + # Ensure lamp and magnet are off before disconnect + await self.send_command([0x03, 0x00]) + await self.send_command([0x0F, 0x00]) + self.dev.reset() + usb.util.dispose_resources(self.dev) + except Exception: + pass + print("NanoDrop safely disconnected.") + + async def send_command(self, payload: List[int]): + """Generic transport method for writing to the command mailbox.""" + self.dev.write(self.EP_OUT, payload) + + async def read_comm(self, timeout=500) -> bytes: + """Reads from the 64-byte text/status endpoint.""" + return self.dev.read(self.EP_IN_COMM, 64, timeout=timeout) + + 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): + data_buffer.extend(self.dev.read(self.EP_IN_HEAVY, 64, timeout=timeout)) + return data_buffer + + def flush_comm(self): + try: + while True: + self.dev.read(self.EP_IN_COMM, 64, timeout=50) + except usb.core.USBTimeoutError: + pass + + def flush_heavy(self): + try: + while True: + self.dev.read(self.EP_IN_HEAVY, 512, timeout=50) + except usb.core.USBTimeoutError: + pass + + +class ThermoFisherNanoDrop1000AbsorbanceBackend( + CapabilityBackend +): # Ideally inherits from AbsorbanceBackend + """ + Translates scientific workflow methods into raw driver commands. + Holds state for coefficients, dark spectra, and blank spectra. + """ + + def __init__(self, driver: ThermoFisherNanoDrop1000Driver): + super().__init__() + self.driver = driver + self.coefficients = {} + self.wavelengths = None + + self.dark_spectrum = None + self.blank_spectrum = None + + async def _on_setup(self): + """Lifecycle hook to download factory calibration on boot.""" + await self._download_all_coefficients() + self._calculate_x_axis() + print("NanoDrop Initialized and Calibrated.") + + async def _on_stop(self): + """Lifecycle hook to clean up state on teardown.""" + self.coefficients.clear() + + async def set_lamp(self, state: bool): + cmd = 0xFF if state else 0x00 + await self.driver.send_command([0x03, cmd]) + + async def set_magnet(self, state: bool): + cmd = 0xFF if state else 0x00 + await self.driver.send_command([0x0F, cmd]) + + async def set_integration_time(self, ms: int): + if ms < 3: + ms = 3 + print("Integration too low, setting to 3 ms") + elif ms > 65535: + ms = 65535 + print("Integration too high, setting to 65535 ms") + + lsb = ms & 0xFF + msb = (ms >> 8) & 0xFF + await self.driver.send_command([0x02, lsb, msb]) + + async def _download_all_coefficients(self): + print("Downloading Factory Memory Map...") + self.driver.flush_comm() + + for index in range(1, 15): + if index == 5: + continue + await self.driver.send_command([0x05, index]) + await asyncio.sleep(0.05) + try: + data = await self.driver.read_comm() + text = bytearray(data[2:]).decode("ascii", errors="ignore").split("\x00")[0] + self.coefficients[index] = float(text) + except Exception: + print(f"Warning: Failed to read coefficient index {index}") + + def _calculate_x_axis(self): + pixels = np.arange(2048) + c0, c1 = self.coefficients.get(1, 0), self.coefficients.get(2, 0) + c2, c3 = self.coefficients.get(3, 0), self.coefficients.get(4, 0) + self.wavelengths = c0 + (c1 * pixels) + (c2 * (pixels**2)) + (c3 * (pixels**3)) + + async def get_raw_spectrum(self) -> np.ndarray: + self.driver.flush_heavy() + await self.driver.send_command([0x09]) + + data_buffer = await self.driver.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 = np.array(pixels, dtype=float) + + # 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) + print("Acquiring Dark baseline...") + self.dark_spectrum = await self.get_raw_spectrum() + + await self.set_lamp(True) + await asyncio.sleep(0.2) + print("Acquiring Blank baseline...") + self.blank_spectrum = await self.get_raw_spectrum() + + await self.set_lamp(False) + await self.set_magnet(False) + print("Blanking complete.") + + async def measure_absorbance(self, integration_ms=20) -> Tuple[np.ndarray, np.ndarray]: + 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) + + print("Measuring sample...") + sample_spectrum = await self.get_raw_spectrum() + + await self.set_lamp(False) + await self.set_magnet(False) + + numerator = np.clip(sample_spectrum - self.dark_spectrum, 1, None) + denominator = np.clip(self.blank_spectrum - self.dark_spectrum, 1, None) + + transmittance = numerator / denominator + absorbance = -np.log10(transmittance) + + return self.wavelengths, absorbance class ThermoFisherNanoDrop1000(Device): - """ - Main PyLabRobot Device Class. - Constructs the driver and registers the absorbance capability. - """ - def __init__(self, name: str = "NanoDrop1000"): - super().__init__(name=name) - - # Construct ONE driver - self.driver = ThermoFisherNanoDrop1000Driver() - - # Construct backends sharing the single driver - self.absorbance_backend = ThermoFisherNanoDrop1000AbsorbanceBackend(driver=self.driver) - - # Append to capabilities - self._capabilities.append(self.absorbance_backend) - - # setup() and stop() are completely removed! Inherited behavior takes over. - - # CAUTION: Convenience methods below map to Capability operations. - async def take_blank(self, integration_ms=20): - await self.absorbance_backend.take_blank(integration_ms) - - async def measure_absorbance(self, integration_ms=20): - return await self.absorbance_backend.measure_absorbance(integration_ms) \ No newline at end of file + """ + Main PyLabRobot Device Class. + Constructs the driver and registers the absorbance capability. + """ + + def __init__(self, name: str = "NanoDrop1000"): + super().__init__(name=name) + + # Construct ONE driver + self.driver = ThermoFisherNanoDrop1000Driver() + + # Construct backends sharing the single driver + self.absorbance_backend = ThermoFisherNanoDrop1000AbsorbanceBackend(driver=self.driver) + + # Append to capabilities + self._capabilities.append(self.absorbance_backend) + + # setup() and stop() are completely removed! Inherited behavior takes over. + + # CAUTION: Convenience methods below map to Capability operations. + async def take_blank(self, integration_ms=20): + await self.absorbance_backend.take_blank(integration_ms) + + async def measure_absorbance(self, integration_ms=20): + return await self.absorbance_backend.measure_absorbance(integration_ms) From a371dde0dbb05d60b191bc2817f747608f77ab0a Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Sat, 15 Aug 2026 20:48:22 -0700 Subject: [PATCH 3/8] simplify file --- .../ThermoFisherNanoDrop1000_PLR_V1.py | 96 ++++--------------- 1 file changed, 20 insertions(+), 76 deletions(-) diff --git a/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py index 30d8ac13be5..8af3db7e4cf 100644 --- a/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py +++ b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py @@ -5,19 +5,9 @@ import usb.core import usb.util -from pylabrobot.capabilities import CapabilityBackend -# PyLabRobot core imports (Assuming standard PLR v1b1 structure) -from pylabrobot.device import Device, Driver - -# Assuming an Absorbance Capability exists in PLR; if not, you would define it in capabilities/ -# from pylabrobot.capabilities.absorbance import AbsorbanceBackend, Absorbance - - -class ThermoFisherNanoDrop1000Driver(Driver): +class ThermoFisherNanoDrop1000: """ - Pure transport layer for the NanoDrop 1000. - Handles USB connections, endpoints, and raw byte transfers. """ VID = 0x2457 @@ -30,6 +20,11 @@ def __init__(self): super().__init__() self.dev = None + self.coefficients = {} + self.wavelengths = None + self.dark_spectrum = None + self.blank_spectrum = None + async def setup(self): """Initializes the USB connection.""" print("Connecting to NanoDrop...") @@ -48,6 +43,9 @@ async def setup(self): await self.send_command([0x01]) await asyncio.sleep(0.2) + await self._download_all_coefficients() + self._calculate_x_axis() + async def stop(self): """Safely powers down hardware and releases the USB port.""" if self.dev: @@ -61,6 +59,8 @@ async def stop(self): pass print("NanoDrop safely disconnected.") + self.coefficients = {} + async def send_command(self, payload: List[int]): """Generic transport method for writing to the command mailbox.""" self.dev.write(self.EP_OUT, payload) @@ -90,41 +90,13 @@ def flush_heavy(self): except usb.core.USBTimeoutError: pass - -class ThermoFisherNanoDrop1000AbsorbanceBackend( - CapabilityBackend -): # Ideally inherits from AbsorbanceBackend - """ - Translates scientific workflow methods into raw driver commands. - Holds state for coefficients, dark spectra, and blank spectra. - """ - - def __init__(self, driver: ThermoFisherNanoDrop1000Driver): - super().__init__() - self.driver = driver - self.coefficients = {} - self.wavelengths = None - - self.dark_spectrum = None - self.blank_spectrum = None - - async def _on_setup(self): - """Lifecycle hook to download factory calibration on boot.""" - await self._download_all_coefficients() - self._calculate_x_axis() - print("NanoDrop Initialized and Calibrated.") - - async def _on_stop(self): - """Lifecycle hook to clean up state on teardown.""" - self.coefficients.clear() - async def set_lamp(self, state: bool): cmd = 0xFF if state else 0x00 - await self.driver.send_command([0x03, cmd]) + await self.send_command([0x03, cmd]) async def set_magnet(self, state: bool): cmd = 0xFF if state else 0x00 - await self.driver.send_command([0x0F, cmd]) + await self.send_command([0x0F, cmd]) async def set_integration_time(self, ms: int): if ms < 3: @@ -136,19 +108,19 @@ async def set_integration_time(self, ms: int): lsb = ms & 0xFF msb = (ms >> 8) & 0xFF - await self.driver.send_command([0x02, lsb, msb]) + await self.send_command([0x02, lsb, msb]) async def _download_all_coefficients(self): print("Downloading Factory Memory Map...") - self.driver.flush_comm() + self.flush_comm() for index in range(1, 15): if index == 5: continue - await self.driver.send_command([0x05, index]) + await self.send_command([0x05, index]) await asyncio.sleep(0.05) try: - data = await self.driver.read_comm() + data = await self.read_comm() text = bytearray(data[2:]).decode("ascii", errors="ignore").split("\x00")[0] self.coefficients[index] = float(text) except Exception: @@ -161,10 +133,10 @@ def _calculate_x_axis(self): self.wavelengths = c0 + (c1 * pixels) + (c2 * (pixels**2)) + (c3 * (pixels**3)) async def get_raw_spectrum(self) -> np.ndarray: - self.driver.flush_heavy() - await self.driver.send_command([0x09]) + self.flush_heavy() + await self.send_command([0x09]) - data_buffer = await self.driver.read_heavy() + data_buffer = await self.read_heavy() pixels = [] for i in range(0, 4096, 128): @@ -229,31 +201,3 @@ async def measure_absorbance(self, integration_ms=20) -> Tuple[np.ndarray, np.nd absorbance = -np.log10(transmittance) return self.wavelengths, absorbance - - -class ThermoFisherNanoDrop1000(Device): - """ - Main PyLabRobot Device Class. - Constructs the driver and registers the absorbance capability. - """ - - def __init__(self, name: str = "NanoDrop1000"): - super().__init__(name=name) - - # Construct ONE driver - self.driver = ThermoFisherNanoDrop1000Driver() - - # Construct backends sharing the single driver - self.absorbance_backend = ThermoFisherNanoDrop1000AbsorbanceBackend(driver=self.driver) - - # Append to capabilities - self._capabilities.append(self.absorbance_backend) - - # setup() and stop() are completely removed! Inherited behavior takes over. - - # CAUTION: Convenience methods below map to Capability operations. - async def take_blank(self, integration_ms=20): - await self.absorbance_backend.take_blank(integration_ms) - - async def measure_absorbance(self, integration_ms=20): - return await self.absorbance_backend.measure_absorbance(integration_ms) From 708356e0ecd5e3cde74f75676e69c72a7c03e850 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Sat, 15 Aug 2026 20:57:01 -0700 Subject: [PATCH 4/8] Use PLR USB for NanoDrop 1000 --- docs/user_guide/thermo_fisher/index.md | 2 + .../nanodrop_1000/hello-world.ipynb | 79 ++++++++++ .../thermo_fisher/nanodrop_1000/setup.md | 49 ++++++ .../ThermoFisherNanoDrop1000_PLR_V1.py | 142 +++++++++++------- pylabrobot/thermo_fisher/bosdescriptor.reg | 3 - .../thermo_fisher/nanodrop_setup_guide.md | 88 ----------- 6 files changed, 214 insertions(+), 149 deletions(-) create mode 100644 docs/user_guide/thermo_fisher/nanodrop_1000/hello-world.ipynb create mode 100644 docs/user_guide/thermo_fisher/nanodrop_1000/setup.md delete mode 100644 pylabrobot/thermo_fisher/bosdescriptor.reg delete mode 100644 pylabrobot/thermo_fisher/nanodrop_setup_guide.md diff --git a/docs/user_guide/thermo_fisher/index.md b/docs/user_guide/thermo_fisher/index.md index 4c887655bbb..6b99610e7fa 100644 --- a/docs/user_guide/thermo_fisher/index.md +++ b/docs/user_guide/thermo_fisher/index.md @@ -4,4 +4,6 @@ :maxdepth: 1 alps/index +nanodrop_1000/hello-world +nanodrop_1000/setup ``` 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..ed5d8bbc672 --- /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.ThermoFisherNanoDrop1000_PLR_V1 import (\n ThermoFisherNanoDrop1000,\n)\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/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/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py index 8af3db7e4cf..78a91baf9b4 100644 --- a/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py +++ b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py @@ -1,41 +1,50 @@ import asyncio -from typing import List, Tuple +import logging +import math +from typing import List, Optional, Tuple -import numpy as np -import usb.core -import usb.util +from pylabrobot.io.usb import USB +logger = logging.getLogger(__name__) -class ThermoFisherNanoDrop1000: - """ - """ +class ThermoFisherNanoDrop1000: VID = 0x2457 PID = 0x1002 EP_OUT = 0x02 EP_IN_HEAVY = 0x82 EP_IN_COMM = 0x87 - def __init__(self): - super().__init__() - self.dev = None + def __init__(self, io: Optional[USB] = None): + self.io = io or 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.wavelengths = None 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.""" - print("Connecting to NanoDrop...") - self.dev = usb.core.find(idVendor=self.VID, idProduct=self.PID) - if self.dev is None: - raise RuntimeError("NanoDrop not found. Is Zadig set to libusb-win32?") - - self.dev.set_configuration() - self.dev.clear_halt(self.EP_OUT) - self.dev.clear_halt(self.EP_IN_HEAVY) - self.dev.clear_halt(self.EP_IN_COMM) + logger.info("Connecting to NanoDrop 1000") + await self.io.setup(empty_buffer=False) + self._connected = True # Wake & Init await self.send_command([0x08]) @@ -48,46 +57,59 @@ async def setup(self): async def stop(self): """Safely powers down hardware and releases the USB port.""" - if self.dev: + if self._connected: try: # Ensure lamp and magnet are off before disconnect await self.send_command([0x03, 0x00]) await self.send_command([0x0F, 0x00]) - self.dev.reset() - usb.util.dispose_resources(self.dev) except Exception: - pass - print("NanoDrop safely disconnected.") + 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.""" - self.dev.write(self.EP_OUT, payload) + await self.io.write(bytes(payload)) async def read_comm(self, timeout=500) -> bytes: """Reads from the 64-byte text/status endpoint.""" - return self.dev.read(self.EP_IN_COMM, 64, timeout=timeout) + 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): - data_buffer.extend(self.dev.read(self.EP_IN_HEAVY, 64, timeout=timeout)) + packet = await asyncio.to_thread( + self.io._read_packet, + 64, + timeout / 1000, + self.EP_IN_HEAVY, + ) + if packet is None: + raise TimeoutError("Timed out reading a NanoDrop spectrum packet") + data_buffer.extend(packet) return data_buffer - def flush_comm(self): + async def flush_comm(self): try: while True: - self.dev.read(self.EP_IN_COMM, 64, timeout=50) - except usb.core.USBTimeoutError: + await self.io.read(timeout=0.05, size=64) + except TimeoutError: pass - def flush_heavy(self): - try: - while True: - self.dev.read(self.EP_IN_HEAVY, 512, timeout=50) - except usb.core.USBTimeoutError: + async def flush_heavy(self): + while ( + await asyncio.to_thread( + self.io._read_packet, + 512, + 0.05, + self.EP_IN_HEAVY, + ) + is not None + ): pass async def set_lamp(self, state: bool): @@ -98,21 +120,21 @@ 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): + async def _set_integration_time(self, ms: int): if ms < 3: ms = 3 - print("Integration too low, setting to 3 ms") + logger.warning("Integration time is too low; using 3 ms") elif ms > 65535: ms = 65535 - print("Integration too high, setting to 65535 ms") + 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): - print("Downloading Factory Memory Map...") - self.flush_comm() + logger.info("Downloading NanoDrop factory memory map") + await self.flush_comm() for index in range(1, 15): if index == 5: @@ -124,16 +146,17 @@ async def _download_all_coefficients(self): text = bytearray(data[2:]).decode("ascii", errors="ignore").split("\x00")[0] self.coefficients[index] = float(text) except Exception: - print(f"Warning: Failed to read coefficient index {index}") + logger.warning("Failed to read coefficient index %d", index, exc_info=True) def _calculate_x_axis(self): - pixels = np.arange(2048) c0, c1 = self.coefficients.get(1, 0), self.coefficients.get(2, 0) c2, c3 = self.coefficients.get(3, 0), self.coefficients.get(4, 0) - self.wavelengths = c0 + (c1 * pixels) + (c2 * (pixels**2)) + (c3 * (pixels**3)) + self.wavelengths = [ + c0 + (c1 * pixel) + (c2 * (pixel**2)) + (c3 * (pixel**3)) for pixel in range(2048) + ] - async def get_raw_spectrum(self) -> np.ndarray: - self.flush_heavy() + async def get_raw_spectrum(self) -> List[float]: + await self.flush_heavy() await self.send_command([0x09]) data_buffer = await self.read_heavy() @@ -145,7 +168,7 @@ async def get_raw_spectrum(self) -> np.ndarray: for j in range(64): pixels.append((msb_block[j] << 8) | lsb_block[j]) - raw_intensities = np.array(pixels, dtype=float) + 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 @@ -158,24 +181,24 @@ async def get_raw_spectrum(self) -> np.ndarray: return raw_intensities async def take_blank(self, integration_ms=20): - await self.set_integration_time(integration_ms) + await self._set_integration_time(integration_ms) await self.set_lamp(False) await self.set_magnet(True) await asyncio.sleep(0.2) - print("Acquiring Dark baseline...") + logger.info("Acquiring dark baseline") self.dark_spectrum = await self.get_raw_spectrum() await self.set_lamp(True) await asyncio.sleep(0.2) - print("Acquiring Blank baseline...") + logger.info("Acquiring blank baseline") self.blank_spectrum = await self.get_raw_spectrum() await self.set_lamp(False) await self.set_magnet(False) - print("Blanking complete.") + logger.info("Blanking complete") - async def measure_absorbance(self, integration_ms=20) -> Tuple[np.ndarray, np.ndarray]: + 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!") @@ -183,21 +206,24 @@ async def measure_absorbance(self, integration_ms=20) -> Tuple[np.ndarray, np.nd # 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_integration_time(integration_ms) await self.set_magnet(True) await self.set_lamp(True) await asyncio.sleep(0.2) - print("Measuring sample...") + logger.info("Measuring sample") sample_spectrum = await self.get_raw_spectrum() await self.set_lamp(False) await self.set_magnet(False) - numerator = np.clip(sample_spectrum - self.dark_spectrum, 1, None) - denominator = np.clip(self.blank_spectrum - self.dark_spectrum, 1, None) - - transmittance = numerator / denominator - absorbance = -np.log10(transmittance) + 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.wavelengths, absorbance diff --git a/pylabrobot/thermo_fisher/bosdescriptor.reg b/pylabrobot/thermo_fisher/bosdescriptor.reg deleted file mode 100644 index ab4d5c453ba..00000000000 --- a/pylabrobot/thermo_fisher/bosdescriptor.reg +++ /dev/null @@ -1,3 +0,0 @@ -Windows Registry Editor Version 5.00 -[HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\usbflags\245710020002] -"SkipBOSDescriptorQuery"=dword:00000001 \ No newline at end of file diff --git a/pylabrobot/thermo_fisher/nanodrop_setup_guide.md b/pylabrobot/thermo_fisher/nanodrop_setup_guide.md deleted file mode 100644 index 8f40bc74b3a..00000000000 --- a/pylabrobot/thermo_fisher/nanodrop_setup_guide.md +++ /dev/null @@ -1,88 +0,0 @@ -``` -The NanoDrop Open-Source Installation Guide -``` - -``` -Step 1: The BOS Descriptor Fix (Windows 10/11) Because the NanoDrop uses an -older USB 1.1 microcontroller, plugging it into a modern Windows 10/11 system -can cause Windows to request a "BOS Descriptor"—a feature that didn't exist when -the machine was built. This causes Windows to flag it as an "Unknown Device." -1.Download the SkipBOSDescriptor.reg file from the GitHub repository. -``` - -`2. Double-click the .reg file and click Yes on the Administrator prompt to merge it into your registry.` - -``` -Important Note: If Windows opens the file in Notepad instead of running -it, Windows is likely hiding file extensions and saved it as a .txt file. To -fix this: Open Windows File Explorer, click the View tab at the top, and check -the box for File name extensions. Rename the downloaded file to ensure it ends -in .reg (not .reg.txt). -``` - -``` -Further, sometimes windows will not let you open .reg files regularly, in -this case you could try: -``` - - `1. Press Win + R to open the Run dialogue box.` - - `2. Type 'regedit' and press Enter to open the Registry Editor.` - - `3. In the top menu, click File > Import.` - - `4. Locate your .reg file, select it, and click Open.` - -`3. Unplug the NanoDrop and plug it back in. It will now be recognized.` - -``` -Step 2: Installing the Python Driver (Zadig) To control the machine with Python, -we must temporarily replace the official driver with an open-source one. -``` - -`1. Download and run Zadig (zadig.akeo.ie).` - -`2. Go to Options -> List All Devices.` - -`3. Select the NanoDrop 1000 from the main dropdown menu.` - -``` -4.On the right side of the green arrow, use the up/down arrows to select -libusb-win32. -``` - -``` -5.Click Replace Driver (or Install Driver) and wait for the "Success" -message. -``` - -`6. Your Python script can now communicate with the hardware.` - -``` -Reverting to the Official NanoDrop Software -``` - -``` -If you wish to revert to the original software, you can seamlessly swap back to -the proprietary driver without uninstalling anything. -``` - -`1. Open Windows Device Manager.` - -`2. Scroll down and find the NanoDrop 1000 (it will likely be under "libusbwin32 devices").` - -`3. Right-click the device and select Update driver.` - -`4. Click Browse my computer for drivers.` - -`5. Click Let me pick from a list of available drivers on my computer.` - -`6. You will see a list containing the driver you just installed (libusbwin32) and the original official driver (often named something like NanoDrop 1000 Spectrometer or Cypress EZ-USB).` - -`7. Select the original official driver and click Next.` - -``` -8.Wait a few seconds for Windows to swap them over. You can now open the -official NanoDrop software. -``` - From f4b96e06d9229c677694e764cb5841af0652a780 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Sat, 15 Aug 2026 21:19:36 -0700 Subject: [PATCH 5/8] Support USB reads from alternate endpoints --- pylabrobot/io/usb.py | 49 +++++++++++++++++-- .../ThermoFisherNanoDrop1000_PLR_V1.py | 32 +++--------- 2 files changed, 54 insertions(+), 27 deletions(-) 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/ThermoFisherNanoDrop1000_PLR_V1.py b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py index 78a91baf9b4..67d1031105c 100644 --- a/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py +++ b/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py @@ -15,8 +15,8 @@ class ThermoFisherNanoDrop1000: EP_IN_HEAVY = 0x82 EP_IN_COMM = 0x87 - def __init__(self, io: Optional[USB] = None): - self.io = io or USB( + def __init__(self): + self.io = USB( id_vendor=self.VID, id_product=self.PID, human_readable_device_name="Thermo Fisher NanoDrop 1000", @@ -82,35 +82,19 @@ 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 asyncio.to_thread( - self.io._read_packet, - 64, - timeout / 1000, - self.EP_IN_HEAVY, + packet = await self.io.read( + timeout=timeout / 1000, + size=64, + endpoint=self.EP_IN_HEAVY, ) - if packet is None: - raise TimeoutError("Timed out reading a NanoDrop spectrum packet") data_buffer.extend(packet) return data_buffer async def flush_comm(self): - try: - while True: - await self.io.read(timeout=0.05, size=64) - except TimeoutError: - pass + await self.io.drain(endpoint=self.EP_IN_COMM, timeout=0.05, size=64) async def flush_heavy(self): - while ( - await asyncio.to_thread( - self.io._read_packet, - 512, - 0.05, - self.EP_IN_HEAVY, - ) - is not None - ): - pass + 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 From 4fd5c7e345933f638d33980ed9cae2dfb92520e3 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Sat, 15 Aug 2026 21:53:26 -0700 Subject: [PATCH 6/8] Organize NanoDrop 1000 support --- docs/user_guide/thermo_fisher/index.md | 3 +-- .../thermo_fisher/nanodrop_1000/hello-world.ipynb | 2 +- docs/user_guide/thermo_fisher/nanodrop_1000/index.md | 12 ++++++++++++ pylabrobot/thermo_fisher/nanodrop_1000/__init__.py | 3 +++ .../nanodrop_1000.py} | 2 +- 5 files changed, 18 insertions(+), 4 deletions(-) create mode 100644 docs/user_guide/thermo_fisher/nanodrop_1000/index.md create mode 100644 pylabrobot/thermo_fisher/nanodrop_1000/__init__.py rename pylabrobot/thermo_fisher/{ThermoFisherNanoDrop1000_PLR_V1.py => nanodrop_1000/nanodrop_1000.py} (99%) diff --git a/docs/user_guide/thermo_fisher/index.md b/docs/user_guide/thermo_fisher/index.md index 6b99610e7fa..f5ea67777ee 100644 --- a/docs/user_guide/thermo_fisher/index.md +++ b/docs/user_guide/thermo_fisher/index.md @@ -4,6 +4,5 @@ :maxdepth: 1 alps/index -nanodrop_1000/hello-world -nanodrop_1000/setup +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 index ed5d8bbc672..f9f481cb4b0 100644 --- a/docs/user_guide/thermo_fisher/nanodrop_1000/hello-world.ipynb +++ b/docs/user_guide/thermo_fisher/nanodrop_1000/hello-world.ipynb @@ -18,7 +18,7 @@ "id": "nanodrop-connect", "metadata": {}, "outputs": [], - "source": "from pylabrobot.thermo_fisher.ThermoFisherNanoDrop1000_PLR_V1 import (\n ThermoFisherNanoDrop1000,\n)\n\nnanodrop = ThermoFisherNanoDrop1000()\nawait nanodrop.setup()" + "source": "from pylabrobot.thermo_fisher.nanodrop_1000 import ThermoFisherNanoDrop1000\n\nnanodrop = ThermoFisherNanoDrop1000()\nawait nanodrop.setup()" }, { "cell_type": "markdown", 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..99820dc8be6 --- /dev/null +++ b/docs/user_guide/thermo_fisher/nanodrop_1000/index.md @@ -0,0 +1,12 @@ +# 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/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/ThermoFisherNanoDrop1000_PLR_V1.py b/pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py similarity index 99% rename from pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py rename to pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py index 67d1031105c..06226c8e461 100644 --- a/pylabrobot/thermo_fisher/ThermoFisherNanoDrop1000_PLR_V1.py +++ b/pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py @@ -1,7 +1,7 @@ import asyncio import logging import math -from typing import List, Optional, Tuple +from typing import List, Tuple from pylabrobot.io.usb import USB From 0a53e949380a4185aa2865bb7705ff15885bd206 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Sat, 15 Aug 2026 21:54:25 -0700 Subject: [PATCH 7/8] Calculate NanoDrop wavelengths on demand --- pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py b/pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py index 06226c8e461..b55c052ff4f 100644 --- a/pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py +++ b/pylabrobot/thermo_fisher/nanodrop_1000/nanodrop_1000.py @@ -29,7 +29,6 @@ def __init__(self): self._connected = False self.coefficients = {} - self.wavelengths = None self.dark_spectrum = None self.blank_spectrum = None @@ -53,7 +52,6 @@ async def setup(self): await asyncio.sleep(0.2) await self._download_all_coefficients() - self._calculate_x_axis() async def stop(self): """Safely powers down hardware and releases the USB port.""" @@ -132,10 +130,10 @@ async def _download_all_coefficients(self): except Exception: logger.warning("Failed to read coefficient index %d", index, exc_info=True) - def _calculate_x_axis(self): + 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) - self.wavelengths = [ + return [ c0 + (c1 * pixel) + (c2 * (pixel**2)) + (c3 * (pixel**3)) for pixel in range(2048) ] @@ -210,4 +208,4 @@ async def measure_absorbance(self, integration_ms=20) -> Tuple[List[float], List ) ] - return self.wavelengths, absorbance + return self._calculate_x_axis(), absorbance From c5b2672698976b5cf0c515cc7b16c9cd872a57f2 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Sat, 15 Aug 2026 22:40:15 -0700 Subject: [PATCH 8/8] Document NanoDrop 1000 device --- docs/_static/devices.json | 14 ++++++++++++++ .../thermo_fisher/nanodrop_1000/index.md | 3 +++ 2 files changed, 17 insertions(+) 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/nanodrop_1000/index.md b/docs/user_guide/thermo_fisher/nanodrop_1000/index.md index 99820dc8be6..acd64af2c04 100644 --- a/docs/user_guide/thermo_fisher/nanodrop_1000/index.md +++ b/docs/user_guide/thermo_fisher/nanodrop_1000/index.md @@ -1,5 +1,8 @@ # 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.