fix possible crash

unilabos/devices/liquid_handling/liquid_handler_abstract.py

@@ -1159,19 +1159,11 @@ 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 == 8 else [0]
+            use_channels = list(range(self.channel_num)) if self.channel_num > 0 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.
@@ -1199,233 +1191,89 @@ 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_targets != 1 and num_sources != 1:
+        if num_sources == 1 and num_targets > 1:
-            if len_asp_vols != num_sources and len_asp_vols != num_targets:
+            # 模式1: 一对多 (1 source -> N 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}")
+            await self._transfer_one_to_many(
-            if len_dis_vols != num_sources and len_dis_vols != num_targets:
+                sources[0],
-                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}")
+                targets,
-
+                tip_racks,
-        if len(use_channels) == 1:
+                use_channels,
-            max_len = max(num_sources, num_targets)
+                asp_vols,
-            for i in range(max_len):
+                dis_vols,
-
+                asp_flow_rates,
-                # 辅助函数：安全地从列表中获取元素，如果列表为空则返回None
+                dis_flow_rates,
-                def safe_get(lst, idx, default=None):
+                offsets,
-                    return [lst[idx]] if lst else default
+                touch_tip,
-
+                liquid_height,
-                # 动态构建参数字典，只传递实际提供的参数
+                blow_out_air_volume,
-                kwargs = {
+                spread,
-                    'sources': [sources[i%num_sources]],
+                mix_stage,
-                    'targets': [targets[i%num_targets]],
+                mix_times,
-                    'tip_racks': tip_racks,
+                mix_vol,
-                    'use_channels': use_channels,
+                mix_rate,
-                    'asp_vols': [asp_vols[i%len_asp_vols]],
+                mix_liquid_height,
-                    'dis_vols': [dis_vols[i%len_dis_vols]],
+                delays,
-                }
+            )
-
+        elif num_sources > 1 and num_targets == 1:
-                # 条件性添加可选参数
+            # 模式2: 多对一 (N sources -> 1 target)
-                if asp_flow_rates is not None:
+            await self._transfer_many_to_one(
-                    kwargs['asp_flow_rates'] = [asp_flow_rates[i%len_asp_vols]]
+                sources,
-                if dis_flow_rates is not None:
+                targets[0],
-                    kwargs['dis_flow_rates'] = [dis_flow_rates[i%len_dis_vols]]
+                tip_racks,
-                if offsets is not None:
+                use_channels,
-                    kwargs['offsets'] = safe_get(offsets, i)
+                asp_vols,
-                if touch_tip is not None:
+                dis_vols,
-                    kwargs['touch_tip'] = touch_tip if touch_tip else False
+                asp_flow_rates,
-                if liquid_height is not None:
+                dis_flow_rates,
-                    kwargs['liquid_height'] = safe_get(liquid_height, i)
+                offsets,
-                if blow_out_air_volume is not None:
+                touch_tip,
-                    kwargs['blow_out_air_volume'] = safe_get(blow_out_air_volume, i)
+                liquid_height,
-                if spread is not None:
+                blow_out_air_volume,
-                    kwargs['spread'] = spread
+                spread,
-                if mix_stage is not None:
+                mix_stage,
-                    kwargs['mix_stage'] = safe_get(mix_stage, i)
+                mix_times,
-                if mix_times is not None:
+                mix_vol,
-                    kwargs['mix_times'] = safe_get(mix_times, i)
+                mix_rate,
-                if mix_vol is not None:
+                mix_liquid_height,
-                    kwargs['mix_vol'] = safe_get(mix_vol, i)
+                delays,
-                if mix_rate is not None:
+            )
-                    kwargs['mix_rate'] = safe_get(mix_rate, i)
+        elif num_sources == num_targets:
-                if mix_liquid_height is not None:
+            # 模式3: 一对一 (N sources -> N targets)
-                    kwargs['mix_liquid_height'] = safe_get(mix_liquid_height, i)
+            await self._transfer_one_to_one(
-                if delays is not None:
+                sources,
-                    kwargs['delays'] = safe_get(delays, i)
+                targets,
-
+                tip_racks,
-                await self._transfer_base_method(**kwargs)
+                use_channels,
-
+                asp_vols,
-    
+                dis_vols,
-
+                asp_flow_rates,
-        # if num_sources == 1 and num_targets > 1:
+                dis_flow_rates,
-        #     # 模式1: 一对多 (1 source -> N targets)
+                offsets,
-        #     await self._transfer_one_to_many(
+                touch_tip,
-        #         sources,
+                liquid_height,
-        #         targets,
+                blow_out_air_volume,
-        #         tip_racks,
+                spread,
-        #         use_channels,
+                mix_stage,
-        #         asp_vols,
+                mix_times,
-        #         dis_vols,
+                mix_vol,
-        #         asp_flow_rates,
+                mix_rate,
-        #         dis_flow_rates,
+                mix_liquid_height,
-        #         offsets,
+                delays,
-        #         touch_tip,
+            )
-        #         liquid_height,
+        else:
-        #         blow_out_air_volume,
+            raise ValueError(
-        #         spread,
+                f"Unsupported transfer mode: {num_sources} sources -> {num_targets} targets. "
-        #         mix_stage,
+                "Supported modes: 1->N, N->1, or N->N."
-        #         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,