From 2e5fac26b3b8041d1da69ff5828163f53527b7e0 Mon Sep 17 00:00:00 2001 From: q434343 <554662886@qq.com> Date: Fri, 13 Feb 2026 13:46:27 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E5=8D=95=E6=9E=AA=E5=A4=B4?= =?UTF-8?q?=E7=9A=84=E5=A4=9A=E5=AF=B9=E5=A4=9A=E7=A7=BB=E6=B6=B2=E5=88=A4?= =?UTF-8?q?=E5=AE=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../liquid_handler_abstract.py | 304 +++++++++++++----- 1 file changed, 228 insertions(+), 76 deletions(-) diff --git a/unilabos/devices/liquid_handling/liquid_handler_abstract.py b/unilabos/devices/liquid_handling/liquid_handler_abstract.py index ec93617..ef69f8d 100644 --- a/unilabos/devices/liquid_handling/liquid_handler_abstract.py +++ b/unilabos/devices/liquid_handling/liquid_handler_abstract.py @@ -1159,11 +1159,19 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware): Number of mix cycles. If *None* (default) no mixing occurs regardless of mix_stage. """ - + num_sources = len(sources) + num_targets = len(targets) + len_asp_vols = len(asp_vols) + len_dis_vols = len(dis_vols) # 确保 use_channels 有默认值 if use_channels is None: # 默认使用设备所有通道（例如 8 通道移液站默认就是 0-7） - use_channels = list(range(self.channel_num)) if self.channel_num > 0 else [0] + use_channels = list(range(self.channel_num)) if self.channel_num == 8 else [0] + elif len(use_channels) == 8: + if self.channel_num != 8: + raise ValueError(f"if channel_num is 8, use_channels length must be 8, but got {len(use_channels)}") + if num_sources%8 != 0 or num_targets%8 != 0 or len_asp_vols%8 != 0 or len_dis_vols%8 != 0: + raise ValueError(f"if channel_num is 8, sources, targets, asp_vols, and dis_vols length must be divisible by 8, but got {num_sources}, {num_targets}, {len_asp_vols}, and {len_dis_vols}") if is_96_well: pass # This mode is not verified. @@ -1191,90 +1199,234 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware): if mix_times is not None: mix_times = int(mix_times) + # 设置tip racks + self.set_tiprack(tip_racks) + # 识别传输模式（mix_times 为 None 也应该能正常移液，只是不做 mix） num_sources = len(sources) num_targets = len(targets) + len_asp_vols = len(asp_vols) + len_dis_vols = len(dis_vols) - if num_sources == 1 and num_targets > 1: - # 模式1: 一对多 (1 source -> N targets) - await self._transfer_one_to_many( - sources[0], - targets, - tip_racks, - use_channels, - asp_vols, - dis_vols, - asp_flow_rates, - dis_flow_rates, - offsets, - touch_tip, - liquid_height, - blow_out_air_volume, - spread, - mix_stage, - mix_times, - mix_vol, - mix_rate, - mix_liquid_height, - delays, - ) - elif num_sources > 1 and num_targets == 1: - # 模式2: 多对一 (N sources -> 1 target) - await self._transfer_many_to_one( - sources, - targets[0], - tip_racks, - use_channels, - asp_vols, - dis_vols, - asp_flow_rates, - dis_flow_rates, - offsets, - touch_tip, - liquid_height, - blow_out_air_volume, - spread, - mix_stage, - mix_times, - mix_vol, - mix_rate, - mix_liquid_height, - delays, - ) - elif num_sources == num_targets: - # 模式3: 一对一 (N sources -> N targets) - await self._transfer_one_to_one( - sources, - targets, - tip_racks, - use_channels, - asp_vols, - dis_vols, - asp_flow_rates, - dis_flow_rates, - offsets, - touch_tip, - liquid_height, - blow_out_air_volume, - spread, - mix_stage, - mix_times, - mix_vol, - mix_rate, - mix_liquid_height, - delays, - ) - else: - raise ValueError( - f"Unsupported transfer mode: {num_sources} sources -> {num_targets} targets. " - "Supported modes: 1->N, N->1, or N->N." - ) + if num_targets != 1 and num_sources != 1: + if len_asp_vols != num_sources and len_asp_vols != num_targets: + raise ValueError(f"asp_vols length must be equal to sources or targets length, but got {len_asp_vols} and {num_sources} and {num_targets}") + if len_dis_vols != num_sources and len_dis_vols != num_targets: + raise ValueError(f"dis_vols length must be equal to sources or targets length, but got {len_dis_vols} and {num_sources} and {num_targets}") + + if len(use_channels) == 1: + max_len = max(num_sources, num_targets) + for i in range(max_len): + + # 辅助函数：安全地从列表中获取元素，如果列表为空则返回None + def safe_get(lst, idx, default=None): + return [lst[idx]] if lst else default + + # 动态构建参数字典，只传递实际提供的参数 + kwargs = { + 'sources': [sources[i%num_sources]], + 'targets': [targets[i%num_targets]], + 'tip_racks': tip_racks, + 'use_channels': use_channels, + 'asp_vols': [asp_vols[i%len_asp_vols]], + 'dis_vols': [dis_vols[i%len_dis_vols]], + } + + # 条件性添加可选参数 + if asp_flow_rates is not None: + kwargs['asp_flow_rates'] = [asp_flow_rates[i%len_asp_vols]] + if dis_flow_rates is not None: + kwargs['dis_flow_rates'] = [dis_flow_rates[i%len_dis_vols]] + if offsets is not None: + kwargs['offsets'] = safe_get(offsets, i) + if touch_tip is not None: + kwargs['touch_tip'] = touch_tip if touch_tip else False + if liquid_height is not None: + kwargs['liquid_height'] = safe_get(liquid_height, i) + if blow_out_air_volume is not None: + kwargs['blow_out_air_volume'] = safe_get(blow_out_air_volume, i) + if spread is not None: + kwargs['spread'] = spread + if mix_stage is not None: + kwargs['mix_stage'] = safe_get(mix_stage, i) + if mix_times is not None: + kwargs['mix_times'] = safe_get(mix_times, i) + if mix_vol is not None: + kwargs['mix_vol'] = safe_get(mix_vol, i) + if mix_rate is not None: + kwargs['mix_rate'] = safe_get(mix_rate, i) + if mix_liquid_height is not None: + kwargs['mix_liquid_height'] = safe_get(mix_liquid_height, i) + if delays is not None: + kwargs['delays'] = safe_get(delays, i) + + await self._transfer_base_method(**kwargs) + + + + # if num_sources == 1 and num_targets > 1: + # # 模式1: 一对多 (1 source -> N targets) + # await self._transfer_one_to_many( + # sources, + # targets, + # tip_racks, + # use_channels, + # asp_vols, + # dis_vols, + # asp_flow_rates, + # dis_flow_rates, + # offsets, + # touch_tip, + # liquid_height, + # blow_out_air_volume, + # spread, + # mix_stage, + # mix_times, + # mix_vol, + # mix_rate, + # mix_liquid_height, + # delays, + # ) + # elif num_sources > 1 and num_targets == 1: + # # 模式2: 多对一 (N sources -> 1 target) + # await self._transfer_many_to_one( + # sources, + # targets[0], + # tip_racks, + # use_channels, + # asp_vols, + # dis_vols, + # asp_flow_rates, + # dis_flow_rates, + # offsets, + # touch_tip, + # liquid_height, + # blow_out_air_volume, + # spread, + # mix_stage, + # mix_times, + # mix_vol, + # mix_rate, + # mix_liquid_height, + # delays, + # ) + # elif num_sources == num_targets: + # # 模式3: 一对一 (N sources -> N targets) + # await self._transfer_one_to_one( + # sources, + # targets, + # tip_racks, + # use_channels, + # asp_vols, + # dis_vols, + # asp_flow_rates, + # dis_flow_rates, + # offsets, + # touch_tip, + # liquid_height, + # blow_out_air_volume, + # spread, + # mix_stage, + # mix_times, + # mix_vol, + # mix_rate, + # mix_liquid_height, + # delays, + # ) + # else: + # raise ValueError( + # f"Unsupported transfer mode: {num_sources} sources -> {num_targets} targets. " + # "Supported modes: 1->N, N->1, or N->N." + # ) return TransferLiquidReturn( sources=ResourceTreeSet.from_plr_resources(list(sources), known_newly_created=False).dump(), # type: ignore targets=ResourceTreeSet.from_plr_resources(list(targets), known_newly_created=False).dump(), # type: ignore ) + async def _transfer_base_method( + self, + sources: Sequence[Container], + targets: Sequence[Container], + tip_racks: Sequence[TipRack], + use_channels: List[int], + asp_vols: List[float], + dis_vols: List[float], + **kwargs + ): + # 从kwargs中提取参数，提供默认值 + asp_flow_rates = kwargs.get('asp_flow_rates') + dis_flow_rates = kwargs.get('dis_flow_rates') + offsets = kwargs.get('offsets') + touch_tip = kwargs.get('touch_tip', False) + liquid_height = kwargs.get('liquid_height') + blow_out_air_volume = kwargs.get('blow_out_air_volume') + spread = kwargs.get('spread', 'wide') + mix_stage = kwargs.get('mix_stage') + mix_times = kwargs.get('mix_times') + mix_vol = kwargs.get('mix_vol') + mix_rate = kwargs.get('mix_rate') + mix_liquid_height = kwargs.get('mix_liquid_height') + delays = kwargs.get('delays') + + tip = [] + tip.extend(next(self.current_tip)) + await self.pick_up_tips(tip) + + if mix_stage in ["before", "both"] and mix_times is not None and mix_times > 0: + await self.mix( + targets=[targets[0]], + mix_time=mix_times, + mix_vol=mix_vol, + offsets=offsets if offsets else None, + height_to_bottom=mix_liquid_height if mix_liquid_height else None, + mix_rate=mix_rate if mix_rate else None, + use_channels=use_channels, + ) + + await self.aspirate( + resources=[sources[0]], + vols=[asp_vols[0]], + use_channels=use_channels, + flow_rates=[asp_flow_rates[0]] if asp_flow_rates and len(asp_flow_rates) > 0 else None, + offsets=[offsets[0]] if offsets and len(offsets) > 0 else None, + liquid_height=[liquid_height[0]] if liquid_height and len(liquid_height) > 0 else None, + blow_out_air_volume=( + [blow_out_air_volume[0]] if blow_out_air_volume and len(blow_out_air_volume) > 0 else None + ), + spread=spread, + ) + if delays is not None: + await self.custom_delay(seconds=delays[0]) + await self.dispense( + resources=[targets[0]], + vols=[dis_vols[0]], + use_channels=use_channels, + flow_rates=[dis_flow_rates[0]] if dis_flow_rates and len(dis_flow_rates) > 0 else None, + offsets=[offsets[0]] if offsets and len(offsets) > 0 else None, + blow_out_air_volume=( + [blow_out_air_volume[0]] if blow_out_air_volume and len(blow_out_air_volume) > 0 else None + ), + liquid_height=[liquid_height[0]] if liquid_height and len(liquid_height) > 0 else None, + spread=spread, + ) + if delays is not None and len(delays) > 1: + await self.custom_delay(seconds=delays[1]) + if mix_stage in ["after", "both"] and mix_times is not None and mix_times > 0: + await self.mix( + targets=[targets[0]], + mix_time=mix_times, + mix_vol=mix_vol, + offsets=offsets if offsets else None, + height_to_bottom=mix_liquid_height if mix_liquid_height else None, + mix_rate=mix_rate if mix_rate else None, + use_channels=use_channels, + ) + if delays is not None and len(delays) > 1: + await self.custom_delay(seconds=delays[0]) + await self.touch_tip(targets[0]) + await self.discard_tips(use_channels=use_channels) + async def _transfer_one_to_one( self, sources: Sequence[Container],