From a948f09f60e960ab84b14903c73eaefa29c39b20 Mon Sep 17 00:00:00 2001 From: Junhan Chang Date: Fri, 30 May 2025 13:33:10 +0800 Subject: [PATCH] add biomek.py demo implementation --- unilabos/devices/liquid_handling/biomek.py | 167 ++++++++++++++++++ .../liquid_handler_abstract.py | 20 ++- 2 files changed, 182 insertions(+), 5 deletions(-) create mode 100644 unilabos/devices/liquid_handling/biomek.py diff --git a/unilabos/devices/liquid_handling/biomek.py b/unilabos/devices/liquid_handling/biomek.py new file mode 100644 index 00000000..c9375e2a --- /dev/null +++ b/unilabos/devices/liquid_handling/biomek.py @@ -0,0 +1,167 @@ +import requests +from typing import List, Sequence, Optional, Union, Literal + +from .liquid_handler_abstract import LiquidHandlerAbstract + + +class LiquidHandlerBiomek(LiquidHandlerAbstract): + """ + Biomek液体处理器的实现类，继承自LiquidHandlerAbstract。 + 该类用于处理Biomek液体处理器的特定操作。 + """ + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._status = "Idle" # 初始状态为 Idle + self._success = False # 初始成功状态为 False + self._status_queue = kwargs.get("status_queue", None) # 状态队列 + self.temp_protocol = {} + + def create_protocol(self, + protocol_name: str, + protocol_description: str, + protocol_version: str, + protocol_author: str, + protocol_date: str, + ): + """ + 创建一个新的协议。 + + Args: + protocol_name (str): 协议名称 + protocol_description (str): 协议描述 + protocol_version (str): 协议版本 + protocol_author (str): 协议作者 + protocol_date (str): 协议日期 + protocol_type (str): 协议类型 + none_keys (List[str]): 需要设置为None的键列表 + + Returns: + dict: 创建的协议字典 + """ + self.temp_protocol = { + "meta": { + "name": protocol_name, + "description": protocol_description, + "version": protocol_version, + "author": protocol_author, + "date": protocol_date, + }, + "labwares": [], + "steps": [], + } + return self.temp_protocol + + def run_protocol(self): + """ + 执行创建的实验流程。 + 工作站的完整执行流程是， + 从 create_protocol 开始，创建新的 method， + 随后执行 transfer_liquid 等操作向实验流程添加步骤， + 最后 run_protocol 执行整个方法。 + + Returns: + dict: 执行结果 + """ + if not self.temp_protocol: + raise ValueError("No protocol created. Please create a protocol first.") + + # 模拟执行协议 + self._status = "Running" + self._success = True + # 在这里可以添加实际执行协议的逻辑 + + response = requests.post("localhost:5000/api/protocols", json=self.temp_protocol) + + def create_resource( + self, + device_id: str, + res_id: str, + class_name: str, + parent: str, + bind_locations: Point, + liquid_input_slot: list[int], + liquid_type: list[str], + liquid_volume: list[int], + slot_on_deck: int, + ): + """ + 创建一个新的资源。 + + Args: + device_id (str): 设备ID + res_id (str): 资源ID + class_name (str): 资源类名 + parent (str): 父级ID + bind_locations (Point): 绑定位置 + liquid_input_slot (list[int]): 液体输入槽列表 + liquid_type (list[str]): 液体类型列表 + liquid_volume (list[int]): 液体体积列表 + slot_on_deck (int): 甲板上的槽位 + + Returns: + dict: 创建的资源字典 + """ + # TODO：需要对好接口，下面这个是临时的 + resource = { + "id": res_id, + "class": class_name, + "parent": parent, + "bind_locations": bind_locations.to_dict(), + "liquid_input_slot": liquid_input_slot, + "liquid_type": liquid_type, + "liquid_volume": liquid_volume, + "slot_on_deck": slot_on_deck, + } + self.temp_protocol["labwares"].append(resource) + return resource + + def transfer_liquid( + self, + sources: Sequence[Container], + targets: Sequence[Container], + tip_racks: Sequence[TipRack], + *, + use_channels: Optional[List[int]] = None, + asp_vols: Union[List[float], float], + dis_vols: Union[List[float], float], + asp_flow_rates: Optional[List[Optional[float]]] = None, + dis_flow_rates: Optional[List[Optional[float]]] = None, + offsets: Optional[List[Coordinate]] = None, + touch_tip: bool = False, + liquid_height: Optional[List[Optional[float]]] = None, + blow_out_air_volume: Optional[List[Optional[float]]] = None, + spread: Literal["wide", "tight", "custom"] = "wide", + is_96_well: bool = False, + mix_stage: Optional[Literal["none", "before", "after", "both"]] = "none", + mix_times: Optional[List(int)] = None, + mix_vol: Optional[int] = None, + mix_rate: Optional[int] = None, + mix_liquid_height: Optional[float] = None, + delays: Optional[List[int]] = None, + none_keys: List[str] = [] + ): + # TODO：需要对好接口，下面这个是临时的 + self.temp_protocol["steps"].append({ + "type": "transfer", + "sources": [source.to_dict() for source in sources], + "targets": [target.to_dict() for target in targets], + "tip_racks": [tip_rack.to_dict() for tip_rack in tip_racks], + "use_channels": use_channels, + "asp_vols": asp_vols, + "dis_vols": dis_vols, + "asp_flow_rates": asp_flow_rates, + "dis_flow_rates": dis_flow_rates, + "offsets": offsets, + "touch_tip": touch_tip, + "liquid_height": liquid_height, + "blow_out_air_volume": blow_out_air_volume, + "spread": spread, + "is_96_well": is_96_well, + "mix_stage": mix_stage, + "mix_times": mix_times, + "mix_vol": mix_vol, + "mix_rate": mix_rate, + "mix_liquid_height": mix_liquid_height, + "delays": delays, + }) \ No newline at end of file diff --git a/unilabos/devices/liquid_handling/liquid_handler_abstract.py b/unilabos/devices/liquid_handling/liquid_handler_abstract.py index c349403e..2596e843 100644 --- a/unilabos/devices/liquid_handling/liquid_handler_abstract.py +++ b/unilabos/devices/liquid_handling/liquid_handler_abstract.py @@ -21,6 +21,18 @@ class LiquidHandlerAbstract(LiquidHandler): # REMOVE LIQUID -------------------------------------------------- # --------------------------------------------------------------- + async def create_protocol(self, + protocol_name: str, + protocol_description: str, + protocol_version: str, + protocol_author: str, + protocol_date: str, + protocol_type: str, + none_keys: List[str] = [] + ): + """Create a new protocol with the given metadata.""" + pass + async def remove_liquid( self, vols: List[float], @@ -158,13 +170,13 @@ class LiquidHandlerAbstract(LiquidHandler): # --------------------------------------------------------------- async def transfer_liquid( self, - asp_vols: Union[List[float], float], - dis_vols: Union[List[float], float], sources: Sequence[Container], targets: Sequence[Container], tip_racks: Sequence[TipRack], *, use_channels: Optional[List[int]] = None, + asp_vols: Union[List[float], float], + dis_vols: Union[List[float], float], asp_flow_rates: Optional[List[Optional[float]]] = None, dis_flow_rates: Optional[List[Optional[float]]] = None, offsets: Optional[List[Coordinate]] = None, @@ -266,9 +278,7 @@ class LiquidHandlerAbstract(LiquidHandler): print(f"Done: {msg}") print(f"Current time: {time.strftime('%H:%M:%S')}") - async def touch_tip(self, - targets: Sequence[Container], - ): + async def touch_tip(self, targets: Sequence[Container]): """Touch the tip to the side of the well.""" await self.aspirate( resources=[targets],