open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2026-02-08 07:55:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev' of https://github.com/dptech-corp/Uni-Lab-OS into dev

Browse Source
This commit is contained in:
Xianwei Qi
2025-12-26 16:25:19 +08:00
parent 394c140830 e6d8d41183
commit bb95c89829
88 changed files with 83331 additions and 15223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6,6 +6,7 @@ import traceback
from collections import Counter
from typing import List, Sequence, Optional, Literal, Union, Iterator, Dict, Any, Callable, Set, cast
from typing_extensions import TypedDict
from pylabrobot.liquid_handling import LiquidHandler, LiquidHandlerBackend, LiquidHandlerChatterboxBackend, Strictness
from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend
from unilabos.devices.liquid_handling.laiyu.backend.laiyu_v_backend import UniLiquidHandlerLaiyuBackend
@@ -28,12 +29,15 @@ from pylabrobot.resources import (
)
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
class SimpleReturn(TypedDict):
samples: list
volumes: list
class LiquidHandlerMiddleware(LiquidHandler):
def __init__(self, backend: LiquidHandlerBackend, deck: Deck, simulator: bool = False, channel_num: int = 8, **kwargs):
self._simulator = simulator
self.channel_num = channel_num
self.pending_liquids_dict = {}
joint_config = kwargs.get("joint_config", None)
if simulator:
if joint_config:
@@ -131,7 +135,9 @@ class LiquidHandlerMiddleware(LiquidHandler):
return await self._simulate_handler.drop_tips(
tip_spots, use_channels, offsets, allow_nonzero_volume, **backend_kwargs
)
return await super().drop_tips(tip_spots, use_channels, offsets, allow_nonzero_volume, **backend_kwargs)
await super().drop_tips(tip_spots, use_channels, offsets, allow_nonzero_volume, **backend_kwargs)
self.pending_liquids_dict = {}
return
async def return_tips(
self, use_channels: Optional[list[int]] = None, allow_nonzero_volume: bool = False, **backend_kwargs
@@ -154,8 +160,10 @@ class LiquidHandlerMiddleware(LiquidHandler):
offsets = [Coordinate.zero()] * len(use_channels)
if self._simulator:
return await self._simulate_handler.discard_tips(use_channels, allow_nonzero_volume, offsets, **backend_kwargs)
return await super().discard_tips(use_channels, allow_nonzero_volume, offsets, **backend_kwargs)
await super().discard_tips(use_channels, allow_nonzero_volume, offsets, **backend_kwargs)
self.pending_liquids_dict = {}
return
def _check_containers(self, resources: Sequence[Resource]):
super()._check_containers(resources)
@@ -171,6 +179,8 @@ class LiquidHandlerMiddleware(LiquidHandler):
spread: Literal["wide", "tight", "custom"] = "wide",
**backend_kwargs,
):
if self._simulator:
return await self._simulate_handler.aspirate(
resources,
@@ -183,7 +193,7 @@ class LiquidHandlerMiddleware(LiquidHandler):
spread,
**backend_kwargs,
)
return await super().aspirate(
await super().aspirate(
resources,
vols,
use_channels,
@@ -195,6 +205,18 @@ class LiquidHandlerMiddleware(LiquidHandler):
**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)
async def dispense(
self,
resources: Sequence[Container],
@@ -206,7 +228,7 @@ class LiquidHandlerMiddleware(LiquidHandler):
blow_out_air_volume: Optional[List[Optional[float]]] = None,
spread: Literal["wide", "tight", "custom"] = "wide",
**backend_kwargs,
):
) -> SimpleReturn:
if self._simulator:
return await self._simulate_handler.dispense(
resources,
@@ -219,7 +241,7 @@ class LiquidHandlerMiddleware(LiquidHandler):
spread,
**backend_kwargs,
)
return await super().dispense(
await super().dispense(
resources,
vols,
use_channels,
@@ -229,7 +251,17 @@ class LiquidHandlerMiddleware(LiquidHandler):
blow_out_air_volume,
**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(
self,
source: Well,
@@ -549,25 +581,66 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
support_touch_tip = True
_ros_node: BaseROS2DeviceNode
def __init__(self, backend: LiquidHandlerBackend, deck: Deck, simulator: bool=False, channel_num:int = 8):
def __init__(self, backend: LiquidHandlerBackend, deck: Deck, simulator: bool=False, channel_num:int = 8, total_height:float = 310):
"""Initialize a LiquidHandler.
Args:
backend: Backend to use.
deck: Deck to use.
"""
backend_type = None
if isinstance(backend, dict) and "type" in backend:
backend_dict = backend.copy()
type_str = backend_dict.pop("type")
try:
# Try to get class from string using globals (current module), or fallback to pylabrobot or unilabos namespaces
backend_cls = None
if type_str in globals():
backend_cls = globals()[type_str]
else:
# Try resolving dotted notation, e.g. "xxx.yyy.ClassName"
components = type_str.split(".")
mod = None
if len(components) > 1:
module_name = ".".join(components[:-1])
try:
import importlib
mod = importlib.import_module(module_name)
except ImportError:
mod = None
if mod is not None:
backend_cls = getattr(mod, components[-1], None)
if backend_cls is None:
# Try pylabrobot style import (if available)
try:
import pylabrobot
backend_cls = getattr(pylabrobot, type_str, None)
except Exception:
backend_cls = None
if backend_cls is not None and isinstance(backend_cls, type):
backend_type = backend_cls(**backend_dict) # pass the rest of dict as kwargs
except Exception as exc:
raise RuntimeError(f"Failed to convert backend type '{type_str}' to class: {exc}")
else:
backend_type = backend
self._simulator = simulator
self.group_info = dict()
super().__init__(backend, deck, simulator, channel_num)
super().__init__(backend_type, deck, simulator, channel_num)
def post_init(self, ros_node: BaseROS2DeviceNode):
self._ros_node = ros_node
@classmethod
def set_liquid(cls, wells: list[Well], liquid_names: list[str], volumes: list[float]):
def set_liquid(cls, wells: list[Well], liquid_names: list[str], volumes: list[float]) -> SimpleReturn:
"""Set the liquid in a well."""
res_samples = []
res_volumes = []
for well, liquid_name, volume in zip(wells, liquid_names, volumes):
well.set_liquids([(liquid_name, volume)]) # type: ignore
res_samples.append({"name": well.name, "sample_uuid": well.unilabos_extra.get("sample_uuid", None)})
res_volumes.append(volume)
return SimpleReturn(samples=res_samples, volumes=res_volumes)
# ---------------------------------------------------------------
# REMOVE LIQUID --------------------------------------------------
# ---------------------------------------------------------------