open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2026-02-04 05:15:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

返回dispense后的sample_uuid

Browse Source
This commit is contained in:
zhangshixiang
2025-12-08 19:32:36 +08:00
parent b46a51c40e
commit d7302c3b35
1 changed files with 24 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
unilabos/devices/liquid_handling/liquid_handler_abstract.py
View File

@@ -37,6 +37,7 @@ class LiquidHandlerMiddleware(LiquidHandler):
def __init__(self, backend: LiquidHandlerBackend, deck: Deck, simulator: bool = False, channel_num: int = 8, **kwargs): def __init__(self, backend: LiquidHandlerBackend, deck: Deck, simulator: bool = False, channel_num: int = 8, **kwargs):
self._simulator = simulator self._simulator = simulator
self.channel_num = channel_num self.channel_num = channel_num
self.pending_liquids_dict = {}
joint_config = kwargs.get("joint_config", None) joint_config = kwargs.get("joint_config", None)
if simulator: if simulator:
if joint_config: if joint_config:
@@ -174,11 +175,7 @@ class LiquidHandlerMiddleware(LiquidHandler):
spread: Literal["wide", "tight", "custom"] = "wide", spread: Literal["wide", "tight", "custom"] = "wide",
**backend_kwargs, **backend_kwargs,
): ):
res_samples = []
res_volumes = []
for resource, volume in zip(resources, vols):
res_samples.append({"name": resource.name, "sample_uuid": resource.unilabos_extra.get("sample_uuid", None)})
res_volumes.append(volume)
if self._simulator: if self._simulator:
return await self._simulate_handler.aspirate( return await self._simulate_handler.aspirate(
@@ -203,6 +200,16 @@ class LiquidHandlerMiddleware(LiquidHandler):
spread, spread,
**backend_kwargs, **backend_kwargs,
) )
res_samples = []
res_volumes = []
for resource, volume, channel in zip(resources, vols, use_channels):
res_samples.append({"name": resource.name, "sample_uuid": resource.unilabos_extra.get("sample_uuid", None)})
res_volumes.append(volume)
self.pending_liquids_dict[channel] = {
"sample_uuid": resource.unilabos_extra.get("sample_uuid", None),
"volume": volume
}
return SimpleReturn(samples=res_samples, volumes=res_volumes) return SimpleReturn(samples=res_samples, volumes=res_volumes)
@@ -217,7 +224,7 @@ class LiquidHandlerMiddleware(LiquidHandler):
blow_out_air_volume: Optional[List[Optional[float]]] = None, blow_out_air_volume: Optional[List[Optional[float]]] = None,
spread: Literal["wide", "tight", "custom"] = "wide", spread: Literal["wide", "tight", "custom"] = "wide",
**backend_kwargs, **backend_kwargs,
): ) -> SimpleReturn:
if self._simulator: if self._simulator:
return await self._simulate_handler.dispense( return await self._simulate_handler.dispense(
resources, resources,
@@ -230,7 +237,7 @@ class LiquidHandlerMiddleware(LiquidHandler):
spread, spread,
**backend_kwargs, **backend_kwargs,
) )
return await super().dispense( await super().dispense(
resources, resources,
vols, vols,
use_channels, use_channels,
@@ -240,7 +247,17 @@ class LiquidHandlerMiddleware(LiquidHandler):
blow_out_air_volume, blow_out_air_volume,
**backend_kwargs, **backend_kwargs,
) )
res_samples = []
res_volumes = []
for resource, volume, channel in zip(resources, vols, use_channels):
res_uuid = self.pending_liquids_dict[channel]["sample_uuid"]
self.pending_liquids_dict[channel]["volume"] -= volume
resource.unilabos_extra["sample_uuid"] = res_uuid
res_samples.append({"name": resource.name, "sample_uuid": res_uuid})
res_volumes.append(volume)
return SimpleReturn(samples=res_samples, volumes=res_volumes)
async def transfer( async def transfer(
self, self,
source: Well, source: Well,