From 8c6c81d3ee44471d799a7f176818b4061561e31e Mon Sep 17 00:00:00 2001 From: Boqiang Tu Date: Thu, 9 Oct 2025 21:37:55 -0700 Subject: [PATCH 1/3] added atc support for proflex backend --- pylabrobot/thermocycling/proflex.py | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/pylabrobot/thermocycling/proflex.py b/pylabrobot/thermocycling/proflex.py index 5767142b7ee..1c7ab37e3e7 100644 --- a/pylabrobot/thermocycling/proflex.py +++ b/pylabrobot/thermocycling/proflex.py @@ -401,8 +401,11 @@ async def _load_num_blocks_and_type(self): elif self.bid == "13": self._num_blocks = 3 self.num_temp_zones = 2 + elif self.bid == "31": + self._num_blocks = 1 + self.num_temp_zones = 1 else: - raise NotImplementedError("Only BID 12 and 13 are supported") + raise NotImplementedError("Only BID 31, 12 and 13 are supported") async def is_block_running(self, block_id: int) -> bool: run_name = await self.get_run_name(block_id=block_id) @@ -554,6 +557,20 @@ async def buzzer_off(self): if self._parse_scpi_response(res)["status"] != "OK": raise ValueError("Failed to turn off buzzer") + async def close_lid(self): + if self.bid != '31': + raise NotImplementedError("Lid control is only available for BID 31 (ATC)") + res=await self.send_command({"cmd": f"lidclose"}, response_timeout=20, read_once=False) + if self._parse_scpi_response(res)["status"] != "OK": + raise ValueError("Failed to close lid") + + async def open_lid(self): + if self.bid != '31': + raise NotImplementedError("Lid control is only available for BID 31 (ATC)") + res=await self.send_command({"cmd": f"lidopen"}, response_timeout=20, read_once=False) + if self._parse_scpi_response(res)["status"] != "OK": + raise ValueError("Failed to open lid") + async def send_morse_code(self, morse_code: str): short_beep_duration = 0.1 long_beep_duration = short_beep_duration * 3 @@ -852,11 +869,7 @@ async def setup( await self.set_block_idle_temp(temp=block_idle_temp, block_id=block_index) await self.set_cover_idle_temp(temp=cover_idle_temp, block_id=block_index) - async def open_lid(self): - raise NotImplementedError("Open lid command is not implemented for Proflex thermocycler") - async def close_lid(self): - raise NotImplementedError("Close lid command is not implemented for Proflex thermocycler") async def deactivate_lid(self, block_id: Optional[int] = None): assert block_id is not None, "block_id must be specified" From e9a4c3b3eb29a72e5d9ee7fbc5e03ec3549a4f61 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Sat, 11 Oct 2025 11:22:51 -0700 Subject: [PATCH 2/3] abstract --- pylabrobot/thermocycling/__init__.py | 15 +-------- .../thermocycling/thermo_fisher/__init__.py | 2 ++ pylabrobot/thermocycling/thermo_fisher/atc.py | 19 +++++++++++ .../thermocycling/thermo_fisher/proflex.py | 11 +++++++ .../{ => thermo_fisher}/proflex_tests.py | 4 +-- .../thermo_fisher_thermocycler.py} | 32 +++++-------------- 6 files changed, 43 insertions(+), 40 deletions(-) create mode 100644 pylabrobot/thermocycling/thermo_fisher/__init__.py create mode 100644 pylabrobot/thermocycling/thermo_fisher/atc.py create mode 100644 pylabrobot/thermocycling/thermo_fisher/proflex.py rename pylabrobot/thermocycling/{ => thermo_fisher}/proflex_tests.py (98%) rename pylabrobot/thermocycling/{proflex.py => thermo_fisher/thermo_fisher_thermocycler.py} (97%) diff --git a/pylabrobot/thermocycling/__init__.py b/pylabrobot/thermocycling/__init__.py index 5499cfc1f28..084f2aed8d7 100644 --- a/pylabrobot/thermocycling/__init__.py +++ b/pylabrobot/thermocycling/__init__.py @@ -1,20 +1,7 @@ -"""This module contains the thermocycling related classes and functions.""" - from .backend import ThermocyclerBackend from .chatterbox import ThermocyclerChatterboxBackend from .opentrons import OpentronsThermocyclerModuleV1, OpentronsThermocyclerModuleV2 from .opentrons_backend import OpentronsThermocyclerBackend -from .proflex import ProflexBackend from .standard import Step +from .thermo_fisher import * from .thermocycler import Thermocycler - -__all__ = [ - "ThermocyclerBackend", - "ThermocyclerChatterboxBackend", - "Thermocycler", - "ProflexBackend", - "Step", - "OpentronsThermocyclerBackend", - "OpentronsThermocyclerModuleV1", - "OpentronsThermocyclerModuleV2", -] diff --git a/pylabrobot/thermocycling/thermo_fisher/__init__.py b/pylabrobot/thermocycling/thermo_fisher/__init__.py new file mode 100644 index 00000000000..43240dfc8f1 --- /dev/null +++ b/pylabrobot/thermocycling/thermo_fisher/__init__.py @@ -0,0 +1,2 @@ +from .atc import ATCBackend +from .proflex import ProflexBackend diff --git a/pylabrobot/thermocycling/thermo_fisher/atc.py b/pylabrobot/thermocycling/thermo_fisher/atc.py new file mode 100644 index 00000000000..5e5fd89faa5 --- /dev/null +++ b/pylabrobot/thermocycling/thermo_fisher/atc.py @@ -0,0 +1,19 @@ +from pylabrobot.thermocycling.thermo_fisher.thermo_fisher_thermocycler import ( + ThermoFisherThermocyclerBackend, +) + + +class ATCBackend(ThermoFisherThermocyclerBackend): + async def close_lid(self): + if self.bid != "31": + raise NotImplementedError("Lid control is only available for BID 31 (ATC)") + res = await self.send_command({"cmd": f"lidclose"}, response_timeout=20, read_once=False) + if self._parse_scpi_response(res)["status"] != "OK": + raise ValueError("Failed to close lid") + + async def open_lid(self): + if self.bid != "31": + raise NotImplementedError("Lid control is only available for BID 31 (ATC)") + res = await self.send_command({"cmd": f"lidopen"}, response_timeout=20, read_once=False) + if self._parse_scpi_response(res)["status"] != "OK": + raise ValueError("Failed to open lid") diff --git a/pylabrobot/thermocycling/thermo_fisher/proflex.py b/pylabrobot/thermocycling/thermo_fisher/proflex.py new file mode 100644 index 00000000000..da6c387e239 --- /dev/null +++ b/pylabrobot/thermocycling/thermo_fisher/proflex.py @@ -0,0 +1,11 @@ +from pylabrobot.thermocycling.thermo_fisher.thermo_fisher_thermocycler import ( + ThermoFisherThermocyclerBackend, +) + + +class ProflexBackend(ThermoFisherThermocyclerBackend): + async def open_lid(self): + raise NotImplementedError("Open lid command is not implemented for Proflex thermocycler") + + async def close_lid(self): + raise NotImplementedError("Close lid command is not implemented for Proflex thermocycler") diff --git a/pylabrobot/thermocycling/proflex_tests.py b/pylabrobot/thermocycling/thermo_fisher/proflex_tests.py similarity index 98% rename from pylabrobot/thermocycling/proflex_tests.py rename to pylabrobot/thermocycling/thermo_fisher/proflex_tests.py index dbc4d703760..8242624ee21 100644 --- a/pylabrobot/thermocycling/proflex_tests.py +++ b/pylabrobot/thermocycling/thermo_fisher/proflex_tests.py @@ -2,8 +2,8 @@ import unittest import unittest.mock -from pylabrobot.thermocycling.proflex import ProflexBackend from pylabrobot.thermocycling.standard import Protocol, Stage, Step +from pylabrobot.thermocycling.thermo_fisher.proflex import ProflexBackend class TestProflexBackend(unittest.IsolatedAsyncioTestCase): @@ -168,7 +168,7 @@ async def test_run_protocol(self): -cover= 105 -mode= Fast -coverEnabled= On - -notes= + -notes= """ ).strip() diff --git a/pylabrobot/thermocycling/proflex.py b/pylabrobot/thermocycling/thermo_fisher/thermo_fisher_thermocycler.py similarity index 97% rename from pylabrobot/thermocycling/proflex.py rename to pylabrobot/thermocycling/thermo_fisher/thermo_fisher_thermocycler.py index 1c7ab37e3e7..32b2c1ed510 100644 --- a/pylabrobot/thermocycling/proflex.py +++ b/pylabrobot/thermocycling/thermo_fisher/thermo_fisher_thermocycler.py @@ -4,16 +4,16 @@ import logging import re import xml.etree.ElementTree as ET +from abc import ABCMeta from base64 import b64decode from dataclasses import dataclass from typing import Any, Dict, List, Optional, cast from xml.dom import minidom from pylabrobot.io import Socket +from pylabrobot.thermocycling.backend import ThermocyclerBackend from pylabrobot.thermocycling.standard import LidStatus, Protocol, Stage, Step -from .backend import ThermocyclerBackend - def _generate_run_info_files( protocol: Protocol, @@ -203,7 +203,7 @@ def stage_to_scpi(stage: Stage, stage_index: int, stage_name_prefix: str) -> dic return data -class ProflexBackend(ThermocyclerBackend): +class ThermoFisherThermocyclerBackend(ThermocyclerBackend, metaclass=ABCMeta): """Backend for Proflex thermocycler.""" def __init__(self, ip: str, port: int = 7000, shared_secret: bytes = b"f4ct0rymt55"): @@ -557,20 +557,6 @@ async def buzzer_off(self): if self._parse_scpi_response(res)["status"] != "OK": raise ValueError("Failed to turn off buzzer") - async def close_lid(self): - if self.bid != '31': - raise NotImplementedError("Lid control is only available for BID 31 (ATC)") - res=await self.send_command({"cmd": f"lidclose"}, response_timeout=20, read_once=False) - if self._parse_scpi_response(res)["status"] != "OK": - raise ValueError("Failed to close lid") - - async def open_lid(self): - if self.bid != '31': - raise NotImplementedError("Lid control is only available for BID 31 (ATC)") - res=await self.send_command({"cmd": f"lidopen"}, response_timeout=20, read_once=False) - if self._parse_scpi_response(res)["status"] != "OK": - raise ValueError("Failed to open lid") - async def send_morse_code(self, morse_code: str): short_beep_duration = 0.1 long_beep_duration = short_beep_duration * 3 @@ -795,7 +781,7 @@ async def get_run_info(self, protocol: Protocol, block_id: int) -> "RunProgress" run_name = await self.get_run_name(block_id=block_id) if not progress: self.logger.info("Protocol completed") - return ProflexBackend.RunProgress( + return ThermoFisherThermocyclerBackend.RunProgress( running=False, stage="completed", elapsed_time=await self.get_elapsed_run_time_from_log(run_name=run_name), @@ -805,7 +791,7 @@ async def get_run_info(self, protocol: Protocol, block_id: int) -> "RunProgress" if progress["RunTitle"] == "-": await self._read_response(timeout=5) self.logger.info("Protocol completed") - return ProflexBackend.RunProgress( + return ThermoFisherThermocyclerBackend.RunProgress( running=False, stage="completed", elapsed_time=await self.get_elapsed_run_time_from_log(run_name=run_name), @@ -814,7 +800,7 @@ async def get_run_info(self, protocol: Protocol, block_id: int) -> "RunProgress" if progress["Stage"] == "POSTRun": self.logger.info("Protocol in POSTRun") - return ProflexBackend.RunProgress( + return ThermoFisherThermocyclerBackend.RunProgress( running=True, stage="POSTRun", elapsed_time=await self.get_elapsed_run_time_from_log(run_name=run_name), @@ -837,7 +823,7 @@ async def get_run_info(self, protocol: Protocol, block_id: int) -> "RunProgress" break await asyncio.sleep(5) self.logger.info("Infinite hold") - return ProflexBackend.RunProgress( + return ThermoFisherThermocyclerBackend.RunProgress( running=False, stage="infinite_hold", elapsed_time=time_elapsed, @@ -846,7 +832,7 @@ async def get_run_info(self, protocol: Protocol, block_id: int) -> "RunProgress" self.logger.info(f"Elapsed time: {time_elapsed}") self.logger.info(f"Remaining time: {remaining_time}") - return ProflexBackend.RunProgress( + return ThermoFisherThermocyclerBackend.RunProgress( running=True, stage=progress["Stage"], elapsed_time=time_elapsed, @@ -869,8 +855,6 @@ async def setup( await self.set_block_idle_temp(temp=block_idle_temp, block_id=block_index) await self.set_cover_idle_temp(temp=cover_idle_temp, block_id=block_index) - - async def deactivate_lid(self, block_id: Optional[int] = None): assert block_id is not None, "block_id must be specified" return await self.set_cover_idle_temp(temp=105, control_enabled=False, block_id=block_id) From 792bd0eb0b423d396b18f7c256975181312a4ac6 Mon Sep 17 00:00:00 2001 From: Rick Wierenga Date: Sat, 11 Oct 2025 11:32:09 -0700 Subject: [PATCH 3/3] x --- pylabrobot/thermocycling/thermo_fisher/atc.py | 4 ++-- pylabrobot/thermocycling/thermo_fisher/proflex_tests.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pylabrobot/thermocycling/thermo_fisher/atc.py b/pylabrobot/thermocycling/thermo_fisher/atc.py index 5e5fd89faa5..55342ebe347 100644 --- a/pylabrobot/thermocycling/thermo_fisher/atc.py +++ b/pylabrobot/thermocycling/thermo_fisher/atc.py @@ -7,13 +7,13 @@ class ATCBackend(ThermoFisherThermocyclerBackend): async def close_lid(self): if self.bid != "31": raise NotImplementedError("Lid control is only available for BID 31 (ATC)") - res = await self.send_command({"cmd": f"lidclose"}, response_timeout=20, read_once=False) + res = await self.send_command({"cmd": "lidclose"}, response_timeout=20, read_once=False) if self._parse_scpi_response(res)["status"] != "OK": raise ValueError("Failed to close lid") async def open_lid(self): if self.bid != "31": raise NotImplementedError("Lid control is only available for BID 31 (ATC)") - res = await self.send_command({"cmd": f"lidopen"}, response_timeout=20, read_once=False) + res = await self.send_command({"cmd": "lidopen"}, response_timeout=20, read_once=False) if self._parse_scpi_response(res)["status"] != "OK": raise ValueError("Failed to open lid") diff --git a/pylabrobot/thermocycling/thermo_fisher/proflex_tests.py b/pylabrobot/thermocycling/thermo_fisher/proflex_tests.py index 8242624ee21..fbf2ab10fd3 100644 --- a/pylabrobot/thermocycling/thermo_fisher/proflex_tests.py +++ b/pylabrobot/thermocycling/thermo_fisher/proflex_tests.py @@ -168,7 +168,7 @@ async def test_run_protocol(self): -cover= 105 -mode= Fast -coverEnabled= On - -notes= + -notes=\x20 """ ).strip()