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

Merge branch 'dev' into hrdev

Browse Source
This commit is contained in:
ZiWei
2025-12-08 20:25:34 +08:00
parent c68903ed83 ed56c1eba2
commit f03ff96ae4
37 changed files with 1628 additions and 622 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
unilabos/app/main.py
View File

@@ -20,6 +20,7 @@ if unilabos_dir not in sys.path:
from unilabos.utils.banner_print import print_status, print_unilab_banner
from unilabos.config.config import load_config, BasicConfig, HTTPConfig
def load_config_from_file(config_path):
if config_path is None:
config_path = os.environ.get("UNILABOS_BASICCONFIG_CONFIG_PATH", None)
@@ -41,7 +42,7 @@ def convert_argv_dashes_to_underscores(args: argparse.ArgumentParser):
for i, arg in enumerate(sys.argv):
for option_string in option_strings:
if arg.startswith(option_string):
new_arg = arg[:2] + arg[2:len(option_string)].replace("-", "_") + arg[len(option_string):]
new_arg = arg[:2] + arg[2 : len(option_string)].replace("-", "_") + arg[len(option_string) :]
sys.argv[i] = new_arg
break
@@ -49,6 +50,8 @@ def convert_argv_dashes_to_underscores(args: argparse.ArgumentParser):
def parse_args():
"""解析命令行参数"""
parser = argparse.ArgumentParser(description="Start Uni-Lab Edge server.")
subparsers = parser.add_subparsers(title="Valid subcommands", dest="command")
parser.add_argument("-g", "--graph", help="Physical setup graph file path.")
parser.add_argument("-c", "--controllers", default=None, help="Controllers config file path.")
parser.add_argument(
@@ -153,6 +156,39 @@ def parse_args():
default=False,
help="Complete registry information",
)
# workflow upload subcommand
workflow_parser = subparsers.add_parser(
"workflow_upload",
aliases=["wf"],
help="Upload workflow from xdl/json/python files",
)
workflow_parser.add_argument(
"-f",
"--workflow_file",
type=str,
required=True,
help="Path to the workflow file (JSON format)",
)
workflow_parser.add_argument(
"-n",
"--workflow_name",
type=str,
default=None,
help="Workflow name, if not provided will use the name from file or filename",
)
workflow_parser.add_argument(
"--tags",
type=str,
nargs="*",
default=[],
help="Tags for the workflow (space-separated)",
)
workflow_parser.add_argument(
"--published",
action="store_true",
default=False,
help="Whether to publish the workflow (default: False)",
)
return parser
@@ -167,7 +203,6 @@ def main():
if not args_dict.get("skip_env_check", False):
from unilabos.utils.environment_check import check_environment
print_status("正在进行环境依赖检查...", "info")
if not check_environment(auto_install=True):
print_status("环境检查失败，程序退出", "error")
os._exit(1)
@@ -239,9 +274,12 @@ def main():
if args_dict.get("sk", ""):
BasicConfig.sk = args_dict.get("sk", "")
print_status("传入了sk参数，优先采用传入参数！", "info")
BasicConfig.working_dir = working_dir
workflow_upload = args_dict.get("command") in ("workflow_upload", "wf")
# 使用远程资源启动
if args_dict["use_remote_resource"]:
if not workflow_upload and args_dict["use_remote_resource"]:
print_status("使用远程资源启动", "info")
from unilabos.app.web import http_client
@@ -254,7 +292,6 @@ def main():
BasicConfig.port = args_dict["port"] if args_dict["port"] else BasicConfig.port
BasicConfig.disable_browser = args_dict["disable_browser"] or BasicConfig.disable_browser
BasicConfig.working_dir = working_dir
BasicConfig.is_host_mode = not args_dict.get("is_slave", False)
BasicConfig.slave_no_host = args_dict.get("slave_no_host", False)
BasicConfig.upload_registry = args_dict.get("upload_registry", False)
@@ -283,9 +320,31 @@ def main():
# 注册表
lab_registry = build_registry(
args_dict["registry_path"], args_dict.get("complete_registry", False), args_dict["upload_registry"]
args_dict["registry_path"], args_dict.get("complete_registry", False), BasicConfig.upload_registry
)
if BasicConfig.upload_registry:
# 设备注册到服务端 - 需要 ak 和 sk
if BasicConfig.ak and BasicConfig.sk:
print_status("开始注册设备到服务端...", "info")
try:
register_devices_and_resources(lab_registry)
print_status("设备注册完成", "info")
except Exception as e:
print_status(f"设备注册失败: {e}", "error")
else:
print_status("未提供 ak 和 sk，跳过设备注册", "info")
else:
print_status("本次启动注册表不报送云端，如果您需要联网调试，请在启动命令增加--upload_registry", "warning")
# 处理 workflow_upload 子命令
if workflow_upload:
from unilabos.workflow.wf_utils import handle_workflow_upload_command
handle_workflow_upload_command(args_dict)
print_status("工作流上传完成，程序退出", "info")
os._exit(0)
if not BasicConfig.ak or not BasicConfig.sk:
print_status("后续运行必须拥有一个实验室，请前往 https://uni-lab.bohrium.com 注册实验室！", "warning")
os._exit(1)
@@ -362,20 +421,6 @@ def main():
args_dict["devices_config"] = resource_tree_set
args_dict["graph"] = graph_res.physical_setup_graph
if BasicConfig.upload_registry:
# 设备注册到服务端 - 需要 ak 和 sk
if BasicConfig.ak and BasicConfig.sk:
print_status("开始注册设备到服务端...", "info")
try:
register_devices_and_resources(lab_registry)
print_status("设备注册完成", "info")
except Exception as e:
print_status(f"设备注册失败: {e}", "error")
else:
print_status("未提供 ak 和 sk，跳过设备注册", "info")
else:
print_status("本次启动注册表不报送云端，如果您需要联网调试，请在启动命令增加--upload_registry", "warning")
if args_dict["controllers"] is not None:
args_dict["controllers_config"] = yaml.safe_load(open(args_dict["controllers"], encoding="utf-8"))
else:
@@ -390,6 +435,7 @@ def main():
comm_client = get_communication_client()
if "websocket" in args_dict["app_bridges"]:
args_dict["bridges"].append(comm_client)
def _exit(signum, frame):
comm_client.stop()
sys.exit(0)
@@ -431,16 +477,13 @@ def main():
resource_visualization.start()
except OSError as e:
if "AMENT_PREFIX_PATH" in str(e):
print_status(
f"ROS 2环境未正确设置，跳过3D可视化启动。错误详情: {e}",
"warning"
)
print_status(f"ROS 2环境未正确设置，跳过3D可视化启动。错误详情: {e}", "warning")
print_status(
"建议解决方案：\n"
"1. 激活Conda环境: conda activate unilab\n"
"2. 或使用 --backend simple 参数\n"
"3. 或使用 --visual disable 参数禁用可视化",
"info"
"info",
)
else:
raise

64
unilabos/app/web/client.py
View File

@@ -76,7 +76,8 @@ class HTTPClient:
Dict[str, str]: 旧UUID到新UUID的映射关系 {old_uuid: new_uuid}
"""
with open(os.path.join(BasicConfig.working_dir, "req_resource_tree_add.json"), "w", encoding="utf-8") as f:
f.write(json.dumps({"nodes": [x for xs in resources.dump() for x in xs], "mount_uuid": mount_uuid}, indent=4))
payload = {"nodes": [x for xs in resources.dump() for x in xs], "mount_uuid": mount_uuid}
f.write(json.dumps(payload, indent=4))
# 从序列化数据中提取所有节点的UUID（保存旧UUID）
old_uuids = {n.res_content.uuid: n for n in resources.all_nodes}
if not self.initialized or first_add:
@@ -331,6 +332,67 @@ class HTTPClient:
logger.error(f"响应内容: {response.text}")
return None
def workflow_import(
self,
name: str,
workflow_uuid: str,
workflow_name: str,
nodes: List[Dict[str, Any]],
edges: List[Dict[str, Any]],
tags: Optional[List[str]] = None,
published: bool = False,
) -> Dict[str, Any]:
"""
导入工作流到服务器
Args:
name: 工作流名称（顶层）
workflow_uuid: 工作流UUID
workflow_name: 工作流名称（data内部）
nodes: 工作流节点列表
edges: 工作流边列表
tags: 工作流标签列表，默认为空列表
published: 是否发布工作流，默认为False
Returns:
Dict: API响应数据，包含 code 和 data (uuid, name)
"""
# target_lab_uuid 暂时使用默认值，后续由后端根据 ak/sk 获取
payload = {
"target_lab_uuid": "28c38bb0-63f6-4352-b0d8-b5b8eb1766d5",
"name": name,
"data": {
"workflow_uuid": workflow_uuid,
"workflow_name": workflow_name,
"nodes": nodes,
"edges": edges,
"tags": tags if tags is not None else [],
"published": published,
},
}
# 保存请求到文件
with open(os.path.join(BasicConfig.working_dir, "req_workflow_upload.json"), "w", encoding="utf-8") as f:
f.write(json.dumps(payload, indent=4, ensure_ascii=False))
response = requests.post(
f"{self.remote_addr}/lab/workflow/owner/import",
json=payload,
headers={"Authorization": f"Lab {self.auth}"},
timeout=60,
)
# 保存响应到文件
with open(os.path.join(BasicConfig.working_dir, "res_workflow_upload.json"), "w", encoding="utf-8") as f:
f.write(f"{response.status_code}" + "\n" + response.text)
if response.status_code == 200:
res = response.json()
if "code" in res and res["code"] != 0:
logger.error(f"导入工作流失败: {response.text}")
return res
else:
logger.error(f"导入工作流失败: {response.status_code}, {response.text}")
return {"code": response.status_code, "message": response.text}
# 创建默认客户端实例
http_client = HTTPClient()

11
unilabos/app/ws_client.py
View File

@@ -438,7 +438,7 @@ class MessageProcessor:
self.connected = True
self.reconnect_count = 0
logger.info(f"[MessageProcessor] Connected to {self.websocket_url}")
logger.trace(f"[MessageProcessor] Connected to {self.websocket_url}")
# 启动发送协程
send_task = asyncio.create_task(self._send_handler())
@@ -503,7 +503,7 @@ class MessageProcessor:
async def _send_handler(self):
"""处理发送队列中的消息"""
logger.debug("[MessageProcessor] Send handler started")
logger.trace("[MessageProcessor] Send handler started")
try:
while self.connected and self.websocket:
@@ -965,7 +965,7 @@ class QueueProcessor:
def _run(self):
"""运行队列处理主循环"""
logger.debug("[QueueProcessor] Queue processor started")
logger.trace("[QueueProcessor] Queue processor started")
while self.is_running:
try:
@@ -1175,7 +1175,6 @@ class WebSocketClient(BaseCommunicationClient):
else:
url = f"{scheme}://{parsed.netloc}/api/v1/ws/schedule"
logger.debug(f"[WebSocketClient] URL: {url}")
return url
def start(self) -> None:
@@ -1188,13 +1187,11 @@ class WebSocketClient(BaseCommunicationClient):
logger.error("[WebSocketClient] WebSocket URL not configured")
return
logger.info(f"[WebSocketClient] Starting connection to {self.websocket_url}")
# 启动两个核心线程
self.message_processor.start()
self.queue_processor.start()
logger.info("[WebSocketClient] All threads started")
logger.trace("[WebSocketClient] All threads started")
def stop(self) -> None:
"""停止WebSocket客户端"""

6
unilabos/config/config.py
View File

@@ -21,7 +21,8 @@ class BasicConfig:
startup_json_path = None # 填写绝对路径
disable_browser = False # 禁止浏览器自动打开
port = 8002 # 本地HTTP服务
log_level: Literal['TRACE', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] = "DEBUG" # 'TRACE', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'
# 'TRACE', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'
log_level: Literal["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "DEBUG"
@classmethod
def auth_secret(cls):
@@ -65,13 +66,14 @@ def _update_config_from_module(module):
if not attr.startswith("_"):
setattr(obj, attr, getattr(getattr(module, name), attr))
def _update_config_from_env():
prefix = "UNILABOS_"
for env_key, env_value in os.environ.items():
if not env_key.startswith(prefix):
continue
try:
key_path = env_key[len(prefix):] # Remove UNILAB_ prefix
key_path = env_key[len(prefix) :] # Remove UNILAB_ prefix
class_field = key_path.upper().split("_", 1)
if len(class_field) != 2:
logger.warning(f"[ENV] 环境变量格式不正确：{env_key}")

29
unilabos/registry/devices/liquid_handler.yaml
View File

@@ -9333,7 +9333,34 @@ liquid_handler.prcxi:
touch_tip: false
use_channels:
- 0
handles: {}
handles:
input:
- data_key: liquid
data_source: handle
data_type: resource
handler_key: sources
label: sources
- data_key: liquid
data_source: executor
data_type: resource
handler_key: targets
label: targets
- data_key: liquid
data_source: executor
data_type: resource
handler_key: tip_rack
label: tip_rack
output:
- data_key: liquid
data_source: handle
data_type: resource
handler_key: sources_out
label: sources
- data_key: liquid
data_source: executor
data_type: resource
handler_key: targets_out
label: targets
placeholder_keys:
sources: unilabos_resources
targets: unilabos_resources

2
unilabos/registry/registry.py
View File

@@ -222,7 +222,7 @@ class Registry:
abs_path = Path(path).absolute()
resource_path = abs_path / "resources"
files = list(resource_path.glob("*/*.yaml"))
logger.debug(f"[UniLab Registry] resources: {resource_path.exists()}, total: {len(files)}")
logger.trace(f"[UniLab Registry] load resources? {resource_path.exists()}, total: {len(files)}")
current_resource_number = len(self.resource_type_registry) + 1
for i, file in enumerate(files):
with open(file, encoding="utf-8", mode="r") as f:

2
unilabos/resources/graphio.py
View File

@@ -42,7 +42,7 @@ def canonicalize_nodes_data(
Returns:
ResourceTreeSet: 标准化后的资源树集合
"""
print_status(f"{len(nodes)} Resources loaded:", "info")
print_status(f"{len(nodes)} Resources loaded", "info")
# 第一步：基本预处理（处理graphml的label字段）
outer_host_node_id = None

4
unilabos/ros/nodes/resource_tracker.py
View File

@@ -66,8 +66,8 @@ class ResourceDict(BaseModel):
klass: str = Field(alias="class", description="Resource class name")
pose: ResourceDictPosition = Field(description="Resource position", default_factory=ResourceDictPosition)
config: Dict[str, Any] = Field(description="Resource configuration")
data: Dict[str, Any] = Field(description="Resource data")
extra: Dict[str, Any] = Field(description="Extra data")
data: Dict[str, Any] = Field(description="Resource data, eg: container liquid data")
extra: Dict[str, Any] = Field(description="Extra data, eg: slot index")
@field_serializer("parent_uuid")
def _serialize_parent(self, parent_uuid: Optional["ResourceDict"]):

181
unilabos/test/resources/bioyond_materials_liquidhandling_1.json
View File

@@ -1,181 +0,0 @@
[
{
"id": "3a1c62c4-c3d2-b803-b72d-7f1153ffef3b",
"typeName": "试剂瓶",
"code": "0004-00050",
"barCode": "",
"name": "NMP",
"quantity": 287.16699029126215,
"lockQuantity": 285.16699029126215,
"unit": "毫升",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14198c-c2d0-efce-0939-69ca5a7dfd39",
"whid": "3a14198c-c2cc-0290-e086-44a428fba248",
"whName": "试剂堆栈",
"code": "0001-0008",
"x": 2,
"y": 4,
"z": 1,
"quantity": 0
}
],
"detail": []
},
{
"id": "3a1cdefe-0e03-1bc1-1296-dae1905c4108",
"typeName": "试剂瓶",
"code": "0004-00052",
"barCode": "",
"name": "NMP",
"quantity": 386.8990291262136,
"lockQuantity": 45.89902912621359,
"unit": "毫升",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14198c-c2d0-f3e7-871a-e470d144296f",
"whid": "3a14198c-c2cc-0290-e086-44a428fba248",
"whName": "试剂堆栈",
"code": "0001-0005",
"x": 2,
"y": 1,
"z": 1,
"quantity": 0
}
],
"detail": []
},
{
"id": "3a1cdefe-0e03-68a4-bcb3-02fc6ba72d1b",
"typeName": "试剂瓶",
"code": "0004-00053",
"barCode": "",
"name": "NMP",
"quantity": 400.0,
"lockQuantity": 0.0,
"unit": "",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14198c-c2d0-2070-efc8-44e245f10c6f",
"whid": "3a14198c-c2cc-0290-e086-44a428fba248",
"whName": "试剂堆栈",
"code": "0001-0006",
"x": 2,
"y": 2,
"z": 1,
"quantity": 0
}
],
"detail": []
},
{
"id": "3a1cdefe-d5e0-d850-5439-4499f20f07fe",
"typeName": "分装板",
"code": "0007-00185",
"barCode": "",
"name": "1010",
"quantity": 1.0,
"lockQuantity": 2.0,
"unit": "块",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14198e-6929-46fe-841e-03dd753f1e4a",
"whid": "3a14198e-6928-121f-7ca6-88ad3ae7e6a0",
"whName": "粉末堆栈",
"code": "0002-0009",
"x": 3,
"y": 1,
"z": 1,
"quantity": 0
}
],
"detail": [
{
"id": "3a1cdefe-d5e0-28a4-f5d0-f7e2436c575f",
"detailMaterialId": "3a1cdefe-d5e0-94ae-f770-27847e73ad38",
"code": null,
"name": "90%分装小瓶",
"quantity": "1",
"lockQuantity": "1",
"unit": "个",
"x": 2,
"y": 3,
"z": 1,
"associateId": null
},
{
"id": "3a1cdefe-d5e0-3ed6-3607-133df89baf5b",
"detailMaterialId": "3a1cdefe-d5e0-f2fa-66bf-94c565d852fb",
"code": null,
"name": "10%分装小瓶",
"quantity": "1",
"lockQuantity": "1",
"unit": "个",
"x": 1,
"y": 3,
"z": 1,
"associateId": null
},
{
"id": "3a1cdefe-d5e0-72b6-e015-be7b93cf09eb",
"detailMaterialId": "3a1cdefe-d5e0-81cf-7dad-2e51cab9ffd6",
"code": null,
"name": "90%分装小瓶",
"quantity": "1",
"lockQuantity": "1",
"unit": "个",
"x": 2,
"y": 1,
"z": 1,
"associateId": null
},
{
"id": "3a1cdefe-d5e0-81d3-ad30-48134afc9ce7",
"detailMaterialId": "3a1cdefe-d5e0-3fa1-cc72-fda6276ae38d",
"code": null,
"name": "10%分装小瓶",
"quantity": "1",
"lockQuantity": "1",
"unit": "个",
"x": 1,
"y": 1,
"z": 1,
"associateId": null
},
{
"id": "3a1cdefe-d5e0-dbdf-d966-9a8926fe1e06",
"detailMaterialId": "3a1cdefe-d5e0-c632-c7da-02d385b18628",
"code": null,
"name": "10%分装小瓶",
"quantity": "1",
"lockQuantity": "1",
"unit": "个",
"x": 1,
"y": 2,
"z": 1,
"associateId": null
},
{
"id": "3a1cdefe-d5e0-f099-b260-e3089a2d08c3",
"detailMaterialId": "3a1cdefe-d5e0-561f-73b6-f8501f814dbb",
"code": null,
"name": "90%分装小瓶",
"quantity": "1",
"lockQuantity": "1",
"unit": "个",
"x": 2,
"y": 2,
"z": 1,
"associateId": null
}
]
}
]

216
unilabos/test/resources/bioyond_materials_liquidhandling_2.json
View File

@@ -1,216 +0,0 @@
[
{
"id": "3a1cde21-a4f4-4f95-6221-eaafc2ae6a8d",
"typeName": "样品瓶",
"code": "0002-00407",
"barCode": "",
"name": "ODA",
"quantity": 25.0,
"lockQuantity": 2.0,
"unit": "克",
"status": 1,
"isUse": false,
"locations": [],
"detail": []
},
{
"id": "3a1cde21-a4f4-7887-9258-e8f8ab7c8a7a",
"typeName": "样品板",
"code": "0008-00160",
"barCode": "",
"name": "1010sample",
"quantity": 1.0,
"lockQuantity": 27.69187,
"unit": "块",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14198e-6929-4379-affa-9a2935c17f99",
"whid": "3a14198e-6928-121f-7ca6-88ad3ae7e6a0",
"whName": "粉末堆栈",
"code": "0002-0002",
"x": 1,
"y": 2,
"z": 1,
"quantity": 0
}
],
"detail": [
{
"id": "3a1cde21-a4f4-0339-f2b6-8e680ad7e8c7",
"detailMaterialId": "3a1cde21-a4f4-ab37-f7a2-ecc3bc083e7c",
"code": null,
"name": "MPDA",
"quantity": "10.505",
"lockQuantity": "-0.0174",
"unit": "克",
"x": 2,
"y": 1,
"z": 1,
"associateId": null
},
{
"id": "3a1cde21-a4f4-a21a-23cf-bb7857b41947",
"detailMaterialId": "3a1cde21-a4f4-99c7-55e7-c80c7320e300",
"code": null,
"name": "ODA",
"quantity": "1.795",
"lockQuantity": "2.0093",
"unit": "克",
"x": 1,
"y": 1,
"z": 1,
"associateId": null
},
{
"id": "3a1cde21-a4f4-af1b-ba0b-2874836800e9",
"detailMaterialId": "3a1cde21-a4f4-4f95-6221-eaafc2ae6a8d",
"code": null,
"name": "ODA",
"quantity": "25",
"lockQuantity": "2",
"unit": "克",
"x": 1,
"y": 2,
"z": 1,
"associateId": null
}
]
},
{
"id": "3a1cde21-a4f4-99c7-55e7-c80c7320e300",
"typeName": "样品瓶",
"code": "0002-00406",
"barCode": "",
"name": "ODA",
"quantity": 1.795,
"lockQuantity": 2.00927,
"unit": "克",
"status": 1,
"isUse": false,
"locations": [],
"detail": []
},
{
"id": "3a1cde21-a4f4-ab37-f7a2-ecc3bc083e7c",
"typeName": "样品瓶",
"code": "0002-00408",
"barCode": "",
"name": "MPDA",
"quantity": 10.505,
"lockQuantity": -0.0174,
"unit": "克",
"status": 1,
"isUse": false,
"locations": [],
"detail": []
},
{
"id": "3a1cdeff-c92a-08f6-c822-732ab734154c",
"typeName": "样品板",
"code": "0008-00161",
"barCode": "",
"name": "1010sample2",
"quantity": 1.0,
"lockQuantity": 3.0,
"unit": "块",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14198e-6929-31f0-8a22-0f98f72260df",
"whid": "3a14198e-6928-121f-7ca6-88ad3ae7e6a0",
"whName": "粉末堆栈",
"code": "0002-0001",
"x": 1,
"y": 1,
"z": 1,
"quantity": 0
}
],
"detail": [
{
"id": "3a1cdeff-c92b-3ace-9623-0bcdef6fa07d",
"detailMaterialId": "3a1cdeff-c92b-d084-2a96-5d62746d9321",
"code": null,
"name": "BTDA1",
"quantity": "0.362",
"lockQuantity": "14.494",
"unit": "克",
"x": 1,
"y": 1,
"z": 1,
"associateId": null
},
{
"id": "3a1cdeff-c92b-856e-f481-792b91b6dbde",
"detailMaterialId": "3a1cdeff-c92b-30f2-f907-8f5e2fe0586b",
"code": null,
"name": "BTDA3",
"quantity": "1.935",
"lockQuantity": "13.067",
"unit": "克",
"x": 1,
"y": 2,
"z": 1,
"associateId": null
},
{
"id": "3a1cdeff-c92b-d144-c5e5-ab9d94e21187",
"detailMaterialId": "3a1cdeff-c92b-519f-a70f-0bb71af537a7",
"code": null,
"name": "BTDA2",
"quantity": "1.903",
"lockQuantity": "13.035",
"unit": "克",
"x": 2,
"y": 1,
"z": 1,
"associateId": null
}
]
},
{
"id": "3a1cdeff-c92b-30f2-f907-8f5e2fe0586b",
"typeName": "样品瓶",
"code": "0002-00411",
"barCode": "",
"name": "BTDA3",
"quantity": 1.935,
"lockQuantity": 13.067,
"unit": "克",
"status": 1,
"isUse": false,
"locations": [],
"detail": []
},
{
"id": "3a1cdeff-c92b-519f-a70f-0bb71af537a7",
"typeName": "样品瓶",
"code": "0002-00410",
"barCode": "",
"name": "BTDA2",
"quantity": 1.903,
"lockQuantity": 13.035,
"unit": "克",
"status": 1,
"isUse": false,
"locations": [],
"detail": []
},
{
"id": "3a1cdeff-c92b-d084-2a96-5d62746d9321",
"typeName": "样品瓶",
"code": "0002-00409",
"barCode": "",
"name": "BTDA1",
"quantity": 0.362,
"lockQuantity": 14.494,
"unit": "克",
"status": 1,
"isUse": false,
"locations": [],
"detail": []
}
]

193
unilabos/test/resources/bioyond_materials_reaction.json
View File

@@ -1,193 +0,0 @@
[
{
"id": "3a1c67a9-aed7-b94d-9e24-bfdf10c8baa9",
"typeName": "烧杯",
"code": "0006-00160",
"barCode": "",
"name": "ODA",
"quantity": 120000.00000000000000000000000,
"lockQuantity": 695374.00000000000000000000000,
"unit": "微升",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14aa17-0d49-11d7-a6e1-f236b3e5e5a3",
"whid": "3a14aa17-0d49-dce4-486e-4b5c85c8b366",
"whName": "堆栈1",
"code": "0001-0001",
"x": 1,
"y": 1,
"z": 1,
"quantity": 0
}
],
"detail": []
},
{
"id": "3a1c67a9-aed9-1ade-5fe1-cc04b24b171c",
"typeName": "烧杯",
"code": "0006-00161",
"barCode": "",
"name": "MPDA",
"quantity": 120000.00000000000000000000000,
"lockQuantity": 681618.00000000000000000000000,
"unit": "",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14aa17-0d49-4bc5-8836-517b75473f5f",
"whid": "3a14aa17-0d49-dce4-486e-4b5c85c8b366",
"whName": "堆栈1",
"code": "0001-0002",
"x": 1,
"y": 2,
"z": 1,
"quantity": 0
}
],
"detail": []
},
{
"id": "3a1c67a9-aed9-2864-6783-2cee4e701ba6",
"typeName": "试剂瓶",
"code": "0004-00041",
"barCode": "",
"name": "NMP",
"quantity": 300000.00000000000000000000000,
"lockQuantity": 380000.00000000000000000000000,
"unit": "微升",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14aa3b-9fab-adac-7b9c-e1ee446b51d5",
"whid": "3a14aa3b-9fab-9d8e-d1a7-828f01f51f0c",
"whName": "站内试剂存放堆栈",
"code": "0003-0001",
"x": 1,
"y": 1,
"z": 1,
"quantity": 0
}
],
"detail": []
},
{
"id": "3a1c67a9-aed9-32c7-5809-3ba1b8db1aa1",
"typeName": "试剂瓶",
"code": "0004-00042",
"barCode": "",
"name": "PGME",
"quantity": 300000.00000000000000000000000,
"lockQuantity": 337892.00000000000000000000000,
"unit": "",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14aa3b-9fab-ca72-febc-b7c304476c78",
"whid": "3a14aa3b-9fab-9d8e-d1a7-828f01f51f0c",
"whName": "站内试剂存放堆栈",
"code": "0003-0002",
"x": 1,
"y": 2,
"z": 1,
"quantity": 0
}
],
"detail": []
},
{
"id": "3a1c68c8-0574-d748-725e-97a2e549f085",
"typeName": "样品板",
"code": "0001-00004",
"barCode": "",
"name": "0917",
"quantity": 1.0000000000000000000000000000,
"lockQuantity": 4.0000000000000000000000000000,
"unit": "块",
"status": 1,
"isUse": false,
"locations": [
{
"id": "3a14aa17-0d49-f49c-6b66-b27f185a3b32",
"whid": "3a14aa17-0d49-dce4-486e-4b5c85c8b366",
"whName": "堆栈1",
"code": "0001-0009",
"x": 2,
"y": 1,
"z": 1,
"quantity": 0
}
],
"detail": [
{
"id": "3a1c68c8-0574-69a1-9858-4637e0193451",
"detailMaterialId": "3a1c68c8-0574-3630-bd42-bbf3623c5208",
"code": null,
"name": "SIDA",
"quantity": "300000",
"lockQuantity": "4",
"unit": "微升",
"x": 1,
"y": 2,
"z": 1,
"associateId": null
},
{
"id": "3a1c68c8-0574-8d51-3191-a31f5be421e5",
"detailMaterialId": "3a1c68c8-0574-3b20-9ad7-90755f123d53",
"code": null,
"name": "BTDA-2",
"quantity": "300000",
"lockQuantity": "4",
"unit": "微升",
"x": 2,
"y": 2,
"z": 1,
"associateId": null
},
{
"id": "3a1c68c8-0574-da80-735b-53ae2197a360",
"detailMaterialId": "3a1c68c8-0574-f2e4-33b3-90d813567939",
"code": null,
"name": "BTDA-DD",
"quantity": "300000",
"lockQuantity": "28",
"unit": "微升",
"x": 1,
"y": 1,
"z": 1,
"associateId": null
},
{
"id": "3a1c68c8-0574-e717-1b1b-99891f875455",
"detailMaterialId": "3a1c68c8-0574-a0ef-e636-68cdc98960e2",
"code": null,
"name": "BTDA-3",
"quantity": "300000",
"lockQuantity": "4",
"unit": "微升",
"x": 2,
"y": 3,
"z": 1,
"associateId": null
},
{
"id": "3a1c68c8-0574-e9bd-6cca-5e261b4f89cb",
"detailMaterialId": "3a1c68c8-0574-9d11-5115-283e8e5510b1",
"code": null,
"name": "BTDA-1",
"quantity": "300000",
"lockQuantity": "4",
"unit": "微升",
"x": 2,
"y": 1,
"z": 1,
"associateId": null
}
]
}
]

48
unilabos/test/resources/test_bottle_carrier.py
View File

@@ -1,48 +0,0 @@
import pytest
from unilabos.resources.bioyond.bottle_carriers import BIOYOND_Electrolyte_6VialCarrier, BIOYOND_Electrolyte_1BottleCarrier
from unilabos.resources.bioyond.bottles import BIOYOND_PolymerStation_Solid_Vial, BIOYOND_PolymerStation_Solution_Beaker, BIOYOND_PolymerStation_Reagent_Bottle
def test_bottle_carrier() -> "BottleCarrier":
print("创建载架...")
# 创建6瓶载架
bottle_carrier = BIOYOND_Electrolyte_6VialCarrier("powder_carrier_01")
print(f"6瓶载架: {bottle_carrier.name}, 位置数: {len(bottle_carrier.sites)}")
# 创建1烧杯载架
beaker_carrier = BIOYOND_Electrolyte_1BottleCarrier("solution_carrier_01")
print(f"1烧杯载架: {beaker_carrier.name}, 位置数: {len(beaker_carrier.sites)}")
# 创建瓶子和烧杯
powder_bottle = BIOYOND_PolymerStation_Solid_Vial("powder_bottle_01")
solution_beaker = BIOYOND_PolymerStation_Solution_Beaker("solution_beaker_01")
reagent_bottle = BIOYOND_PolymerStation_Reagent_Bottle("reagent_bottle_01")
print(f"\n创建的物料:")
print(f"粉末瓶: {powder_bottle.name} - {powder_bottle.diameter}mm x {powder_bottle.height}mm, {powder_bottle.max_volume}μL")
print(f"溶液烧杯: {solution_beaker.name} - {solution_beaker.diameter}mm x {solution_beaker.height}mm, {solution_beaker.max_volume}μL")
print(f"试剂瓶: {reagent_bottle.name} - {reagent_bottle.diameter}mm x {reagent_bottle.height}mm, {reagent_bottle.max_volume}μL")
# 测试放置容器
print(f"\n测试放置容器...")
# 通过载架的索引操作来放置容器
# bottle_carrier[0] = powder_bottle # 放置粉末瓶到第一个位置
print(f"粉末瓶已放置到6瓶载架的位置 0")
# beaker_carrier[0] = solution_beaker # 放置烧杯到第一个位置
print(f"溶液烧杯已放置到1烧杯载架的位置 0")
# 验证放置结果
print(f"\n验证放置结果:")
bottle_at_0 = bottle_carrier[0].resource
beaker_at_0 = beaker_carrier[0].resource
if bottle_at_0:
print(f"位置 0 的瓶子: {bottle_at_0.name}")
if beaker_at_0:
print(f"位置 0 的烧杯: {beaker_at_0.name}")
print("\n载架设置完成！")

66
unilabos/test/resources/test_converter_bioyond.py
View File

@@ -1,66 +0,0 @@
import pytest
import json
import os
from pylabrobot.resources import Resource as ResourcePLR
from unilabos.resources.graphio import resource_bioyond_to_plr
from unilabos.registry.registry import lab_registry
from unilabos.resources.bioyond.decks import BIOYOND_PolymerReactionStation_Deck
lab_registry.setup()
type_mapping = {
"烧杯": ("BIOYOND_PolymerStation_1FlaskCarrier", "3a14196b-24f2-ca49-9081-0cab8021bf1a"),
"试剂瓶": ("BIOYOND_PolymerStation_1BottleCarrier", ""),
"样品板": ("BIOYOND_PolymerStation_6StockCarrier", "3a14196e-b7a0-a5da-1931-35f3000281e9"),
"分装板": ("BIOYOND_PolymerStation_6VialCarrier", "3a14196e-5dfe-6e21-0c79-fe2036d052c4"),
"样品瓶": ("BIOYOND_PolymerStation_Solid_Stock", "3a14196a-cf7d-8aea-48d8-b9662c7dba94"),
"90%分装小瓶": ("BIOYOND_PolymerStation_Solid_Vial", "3a14196c-cdcf-088d-dc7d-5cf38f0ad9ea"),
"10%分装小瓶": ("BIOYOND_PolymerStation_Liquid_Vial", "3a14196c-76be-2279-4e22-7310d69aed68"),
}
@pytest.fixture
def bioyond_materials_reaction() -> list[dict]:
print("加载 BioYond 物料数据...")
print(os.getcwd())
with open("bioyond_materials_reaction.json", "r", encoding="utf-8") as f:
data = json.load(f)
print(f"加载了 {len(data)} 条物料数据")
return data
@pytest.fixture
def bioyond_materials_liquidhandling_1() -> list[dict]:
print("加载 BioYond 物料数据...")
print(os.getcwd())
with open("bioyond_materials_liquidhandling_1.json", "r", encoding="utf-8") as f:
data = json.load(f)
print(f"加载了 {len(data)} 条物料数据")
return data
@pytest.fixture
def bioyond_materials_liquidhandling_2() -> list[dict]:
print("加载 BioYond 物料数据...")
print(os.getcwd())
with open("bioyond_materials_liquidhandling_2.json", "r", encoding="utf-8") as f:
data = json.load(f)
print(f"加载了 {len(data)} 条物料数据")
return data
@pytest.mark.parametrize("materials_fixture", [
"bioyond_materials_reaction",
"bioyond_materials_liquidhandling_1",
])
def test_bioyond_to_plr(materials_fixture, request) -> list[dict]:
materials = request.getfixturevalue(materials_fixture)
deck = BIOYOND_PolymerReactionStation_Deck("test_deck")
output = resource_bioyond_to_plr(materials, type_mapping=type_mapping, deck=deck)
print(deck.summary())
print([resource.serialize() for resource in output])
print([resource.serialize_all_state() for resource in output])
json.dump(deck.serialize(), open("test.json", "w", encoding="utf-8"), indent=4)

115
unilabos/test/resources/test_itemized_carrier.py
View File

@@ -1,115 +0,0 @@
#!/usr/bin/env python3
"""
测试修改后的 get_child_identifier 函数
"""
from unilabos.resources.itemized_carrier import ItemizedCarrier, Bottle
from pylabrobot.resources.coordinate import Coordinate
def test_get_child_identifier_with_indices():
"""测试返回x,y,z索引的 get_child_identifier 函数"""
# 创建一些测试瓶子
bottle1 = Bottle("bottle1", diameter=25.0, height=50.0, max_volume=15.0)
bottle1.location = Coordinate(10, 20, 5)
bottle2 = Bottle("bottle2", diameter=25.0, height=50.0, max_volume=15.0)
bottle2.location = Coordinate(50, 20, 5)
bottle3 = Bottle("bottle3", diameter=25.0, height=50.0, max_volume=15.0)
bottle3.location = Coordinate(90, 20, 5)
# 创建载架，指定维度
sites = {
"A1": bottle1,
"A2": bottle2,
"A3": bottle3,
"B1": None, # 空位
"B2": None,
"B3": None
}
carrier = ItemizedCarrier(
name="test_carrier",
size_x=150,
size_y=100,
size_z=30,
num_items_x=3, # 3列
num_items_y=2, # 2行
num_items_z=1, # 1层
sites=sites
)
print("测试载架维度:")
print(f"num_items_x: {carrier.num_items_x}")
print(f"num_items_y: {carrier.num_items_y}")
print(f"num_items_z: {carrier.num_items_z}")
print()
# 测试获取bottle1的标识符信息 (A1 = idx:0, x:0, y:0, z:0)
result1 = carrier.get_child_identifier(bottle1)
print("测试bottle1 (A1):")
print(f" identifier: {result1['identifier']}")
print(f" idx: {result1['idx']}")
print(f" x index: {result1['x']}")
print(f" y index: {result1['y']}")
print(f" z index: {result1['z']}")
# Assert 验证 bottle1 (A1) 的结果
assert result1['identifier'] == 'A1', f"Expected identifier 'A1', got '{result1['identifier']}'"
assert result1['idx'] == 0, f"Expected idx 0, got {result1['idx']}"
assert result1['x'] == 0, f"Expected x index 0, got {result1['x']}"
assert result1['y'] == 0, f"Expected y index 0, got {result1['y']}"
assert result1['z'] == 0, f"Expected z index 0, got {result1['z']}"
print(" ✓ bottle1 (A1) 测试通过")
print()
# 测试获取bottle2的标识符信息 (A2 = idx:1, x:1, y:0, z:0)
result2 = carrier.get_child_identifier(bottle2)
print("测试bottle2 (A2):")
print(f" identifier: {result2['identifier']}")
print(f" idx: {result2['idx']}")
print(f" x index: {result2['x']}")
print(f" y index: {result2['y']}")
print(f" z index: {result2['z']}")
# Assert 验证 bottle2 (A2) 的结果
assert result2['identifier'] == 'A2', f"Expected identifier 'A2', got '{result2['identifier']}'"
assert result2['idx'] == 1, f"Expected idx 1, got {result2['idx']}"
assert result2['x'] == 1, f"Expected x index 1, got {result2['x']}"
assert result2['y'] == 0, f"Expected y index 0, got {result2['y']}"
assert result2['z'] == 0, f"Expected z index 0, got {result2['z']}"
print(" ✓ bottle2 (A2) 测试通过")
print()
# 测试获取bottle3的标识符信息 (A3 = idx:2, x:2, y:0, z:0)
result3 = carrier.get_child_identifier(bottle3)
print("测试bottle3 (A3):")
print(f" identifier: {result3['identifier']}")
print(f" idx: {result3['idx']}")
print(f" x index: {result3['x']}")
print(f" y index: {result3['y']}")
print(f" z index: {result3['z']}")
# Assert 验证 bottle3 (A3) 的结果
assert result3['identifier'] == 'A3', f"Expected identifier 'A3', got '{result3['identifier']}'"
assert result3['idx'] == 2, f"Expected idx 2, got {result3['idx']}"
assert result3['x'] == 2, f"Expected x index 2, got {result3['x']}"
assert result3['y'] == 0, f"Expected y index 0, got {result3['y']}"
assert result3['z'] == 0, f"Expected z index 0, got {result3['z']}"
print(" ✓ bottle3 (A3) 测试通过")
print()
# 测试错误情况：查找不存在的资源
bottle_not_exists = Bottle("bottle_not_exists", diameter=25.0, height=50.0, max_volume=15.0)
try:
carrier.get_child_identifier(bottle_not_exists)
assert False, "应该抛出 ValueError 异常"
except ValueError as e:
print("✓ 正确抛出了 ValueError 异常：", str(e))
assert "is not assigned to this carrier" in str(e), "异常消息应该包含预期的文本"
print("\n🎉 所有测试都通过了！")
if __name__ == "__main__":
test_get_child_identifier_with_indices()

68
unilabos/test/resources/test_resourcetreeset.py
View File

@@ -1,68 +0,0 @@
import pytest
import json
import os
from pylabrobot.resources import Resource as ResourcePLR
from unilabos.resources.graphio import resource_bioyond_to_plr
from unilabos.ros.nodes.resource_tracker import ResourceTreeSet
from unilabos.registry.registry import lab_registry
from unilabos.resources.bioyond.decks import BIOYOND_PolymerReactionStation_Deck
lab_registry.setup()
type_mapping = {
"烧杯": ("BIOYOND_PolymerStation_1FlaskCarrier", "3a14196b-24f2-ca49-9081-0cab8021bf1a"),
"试剂瓶": ("BIOYOND_PolymerStation_1BottleCarrier", ""),
"样品板": ("BIOYOND_PolymerStation_6StockCarrier", "3a14196e-b7a0-a5da-1931-35f3000281e9"),
"分装板": ("BIOYOND_PolymerStation_6VialCarrier", "3a14196e-5dfe-6e21-0c79-fe2036d052c4"),
"样品瓶": ("BIOYOND_PolymerStation_Solid_Stock", "3a14196a-cf7d-8aea-48d8-b9662c7dba94"),
"90%分装小瓶": ("BIOYOND_PolymerStation_Solid_Vial", "3a14196c-cdcf-088d-dc7d-5cf38f0ad9ea"),
"10%分装小瓶": ("BIOYOND_PolymerStation_Liquid_Vial", "3a14196c-76be-2279-4e22-7310d69aed68"),
}
@pytest.fixture
def bioyond_materials_reaction() -> list[dict]:
print("加载 BioYond 物料数据...")
print(os.getcwd())
with open("bioyond_materials_reaction.json", "r", encoding="utf-8") as f:
data = json.load(f)
print(f"加载了 {len(data)} 条物料数据")
return data
@pytest.fixture
def bioyond_materials_liquidhandling_1() -> list[dict]:
print("加载 BioYond 物料数据...")
print(os.getcwd())
with open("bioyond_materials_liquidhandling_1.json", "r", encoding="utf-8") as f:
data = json.load(f)
print(f"加载了 {len(data)} 条物料数据")
return data
@pytest.fixture
def bioyond_materials_liquidhandling_2() -> list[dict]:
print("加载 BioYond 物料数据...")
print(os.getcwd())
with open("bioyond_materials_liquidhandling_2.json", "r", encoding="utf-8") as f:
data = json.load(f)
print(f"加载了 {len(data)} 条物料数据")
return data
@pytest.mark.parametrize("materials_fixture", [
"bioyond_materials_reaction",
"bioyond_materials_liquidhandling_1",
])
def test_resourcetreeset_from_plr(materials_fixture, request) -> list[dict]:
materials = request.getfixturevalue(materials_fixture)
deck = BIOYOND_PolymerReactionStation_Deck("test_deck")
output = resource_bioyond_to_plr(materials, type_mapping=type_mapping, deck=deck)
print(deck.summary())
r = ResourceTreeSet.from_plr_resources([deck])
print(r.dump())
# json.dump(deck.serialize(), open("test.json", "w", encoding="utf-8"), indent=4)

0
unilabos/test/ros/__init__.py
View File

1
unilabos/test/ros/msgs/__init__.py
View File

@@ -1 +0,0 @@
# 消息转换器测试包

71
unilabos/test/ros/msgs/test_basic.py
View File

@@ -1,71 +0,0 @@
"""
基本测试
测试消息转换器的基本功能，包括导入、类型映射等。
"""
import unittest
from unilabos.ros.msgs.message_converter import (
msg_converter_manager,
get_msg_type,
get_action_type,
get_ros_type_by_msgname,
Point3D,
Point,
Float64,
String,
Bool,
Int32,
)
class TestBasicFunctionality(unittest.TestCase):
"""测试消息转换器的基本功能"""
def test_manager_initialization(self):
"""测试导入管理器初始化"""
self.assertIsNotNone(msg_converter_manager)
self.assertTrue(len(msg_converter_manager.list_modules()) > 0)
self.assertTrue(len(msg_converter_manager.list_classes()) > 0)
def test_get_msg_type(self):
"""测试获取消息类型"""
self.assertEqual(get_msg_type(float), Float64)
self.assertEqual(get_msg_type(str), String)
self.assertEqual(get_msg_type(bool), Bool)
self.assertEqual(get_msg_type(int), Int32)
self.assertEqual(get_msg_type(Point3D), Point)
# 测试错误情况
with self.assertRaises(ValueError):
get_msg_type(set) # 不支持的类型
def test_get_action_type(self):
"""测试获取动作类型"""
float_action = get_action_type(float)
self.assertIsNotNone(float_action)
self.assertTrue("type" in float_action)
self.assertTrue("goal" in float_action)
self.assertTrue("feedback" in float_action)
# 测试错误情况
with self.assertRaises(ValueError):
get_action_type(set) # 不支持的类型
def test_get_ros_type_by_msgname(self):
"""测试通过消息名称获取ROS类型"""
# 测试有效的消息名称
point_type = get_ros_type_by_msgname("geometry_msgs/msg/Point")
self.assertEqual(point_type, Point)
# 测试无效的消息名称
with self.assertRaises(ValueError):
get_ros_type_by_msgname("invalid_format")
# 不存在的消息类型可能会引发ImportError，但这依赖于运行环境
# 因此不进行显式测试
if __name__ == "__main__":
unittest.main()

131
unilabos/test/ros/msgs/test_conversion.py
View File

@@ -1,131 +0,0 @@
"""
转换测试
测试Python对象和ROS消息之间的转换功能。
"""
import unittest
from dataclasses import dataclass
from unilabos.ros.msgs.message_converter import (
convert_to_ros_msg,
convert_from_ros_msg,
convert_to_ros_msg_with_mapping,
convert_from_ros_msg_with_mapping,
Point,
Float64,
String,
Point3D,
Resource,
)
# 定义一些测试数据类
@dataclass
class TestPoint:
x: float = 0.0
y: float = 0.0
z: float = 0.0
class TestBasicConversion(unittest.TestCase):
"""测试基本类型转换"""
def test_primitive_conversion(self):
"""测试原始类型转换"""
# Float转换
float_value = 3.14
ros_float = convert_to_ros_msg(Float64, float_value)
self.assertEqual(ros_float.data, float_value)
# 反向转换
py_float = convert_from_ros_msg(ros_float)
self.assertEqual(py_float, float_value)
# 字符串转换
str_value = "hello"
ros_str = convert_to_ros_msg(String, str_value)
self.assertEqual(ros_str.data, str_value)
# 反向转换
py_str = convert_from_ros_msg(ros_str)
self.assertEqual(py_str, str_value)
def test_point_conversion(self):
"""测试点类型转换"""
# 创建Point3D对象
py_point = Point3D(x=1.0, y=2.0, z=3.0)
# 转换为ROS Point
ros_point = convert_to_ros_msg(Point, py_point)
self.assertEqual(ros_point.x, py_point.x)
self.assertEqual(ros_point.y, py_point.y)
self.assertEqual(ros_point.z, py_point.z)
# 反向转换
py_point_back = convert_from_ros_msg(ros_point)
self.assertEqual(py_point_back.x, py_point.x)
self.assertEqual(py_point_back.y, py_point.y)
self.assertEqual(py_point_back.z, py_point.z)
def test_dataclass_conversion(self):
"""测试dataclass转换"""
# 创建dataclass
test_point = TestPoint(x=1.0, y=2.0, z=3.0)
# 转换
ros_point = convert_to_ros_msg(Point, test_point)
self.assertEqual(ros_point.x, test_point.x)
self.assertEqual(ros_point.y, test_point.y)
self.assertEqual(ros_point.z, test_point.z)
class TestMappingConversion(unittest.TestCase):
"""测试映射转换功能"""
def test_mapping_conversion(self):
"""测试带映射的转换"""
# 创建测试数据
test_data = {
"position": {"x": 1.0, "y": 2.0, "z": 3.0},
"name": "test_resource",
"id": "123",
"type": "test_type",
}
# 定义映射
mapping = {
"id": "id",
"name": "name",
"type": "type",
"pose.position": "position",
}
# 转换为ROS资源
ros_resource = convert_to_ros_msg_with_mapping(Resource, test_data, mapping)
self.assertEqual(ros_resource.id, "123")
self.assertEqual(ros_resource.name, "test_resource")
self.assertEqual(ros_resource.type, "test_type")
self.assertEqual(ros_resource.pose.position.x, 1.0)
self.assertEqual(ros_resource.pose.position.y, 2.0)
self.assertEqual(ros_resource.pose.position.z, 3.0)
# 反向转换
reverse_mapping = {
"id": "id",
"name": "name",
"type": "type",
"pose.position": "position",
}
py_data = convert_from_ros_msg_with_mapping(ros_resource, reverse_mapping)
self.assertEqual(py_data["id"], "123")
self.assertEqual(py_data["name"], "test_resource")
self.assertEqual(py_data["type"], "test_type")
self.assertEqual(py_data["position"].x, 1.0)
self.assertEqual(py_data["position"].y, 2.0)
self.assertEqual(py_data["position"].z, 3.0)
if __name__ == "__main__":
unittest.main()

120
unilabos/test/ros/msgs/test_mapping.py
View File

@@ -1,120 +0,0 @@
"""
映射测试
测试消息类型映射和字段映射功能。
"""
import unittest
from dataclasses import dataclass
from unilabos.ros.msgs.message_converter import (
_msg_mapping,
_action_mapping,
_msg_converter,
_msg_converter_back,
compare_model_fields,
Point,
Point3D,
Float64,
String,
set_msg_data,
)
@dataclass
class TestMappingModel:
"""用于测试映射的数据类"""
id: str
name: str
value: float
@dataclass
class TestPointModel:
"""用于测试字段比较的点模型"""
x: float
y: float
z: float
class TestTypeMapping(unittest.TestCase):
"""测试类型映射"""
def test_msg_mapping(self):
"""测试消息类型映射"""
self.assertIn(float, _msg_mapping)
self.assertEqual(_msg_mapping[float], Float64)
self.assertIn(str, _msg_mapping)
self.assertEqual(_msg_mapping[str], String)
self.assertIn(Point3D, _msg_mapping)
self.assertEqual(_msg_mapping[Point3D], Point)
def test_action_mapping(self):
"""测试动作类型映射"""
self.assertIn(float, _action_mapping)
self.assertIn("type", _action_mapping[float])
self.assertIn("goal", _action_mapping[float])
self.assertIn("feedback", _action_mapping[float])
self.assertIn("result", _action_mapping[float])
def test_converter_mapping(self):
"""测试转换器映射"""
# 测试Python到ROS映射
self.assertIn(float, _msg_converter)
self.assertIn(Float64, _msg_converter)
self.assertIn(String, _msg_converter)
self.assertIn(Point, _msg_converter)
# 测试ROS到Python映射
self.assertIn(float, _msg_converter_back)
self.assertIn(Float64, _msg_converter_back)
self.assertIn(String, _msg_converter_back)
self.assertIn(Point, _msg_converter_back)
class TestFieldMapping(unittest.TestCase):
"""测试字段映射"""
def test_compare_model_fields(self):
"""测试模型字段比较"""
# Point3D和TestPointModel有相同的字段
self.assertTrue(compare_model_fields(Point3D, TestPointModel))
# 与其他类型比较
self.assertFalse(compare_model_fields(Point3D, TestMappingModel))
self.assertFalse(compare_model_fields(Point3D, float))
# 类型对象和实例对象比较
point = Point3D(x=1.0, y=2.0, z=3.0)
self.assertTrue(compare_model_fields(Point3D, type(point)))
def test_set_msg_data(self):
"""测试设置消息数据类型"""
# 测试float转换
float_value = "3.14"
self.assertEqual(set_msg_data("float", float_value), 3.14)
self.assertEqual(set_msg_data("double", float_value), 3.14)
# 测试int转换
int_value = "42"
self.assertEqual(set_msg_data("int", int_value), 42)
# 测试bool转换
bool_value = "True"
self.assertEqual(set_msg_data("bool", bool_value), True)
# 测试str转换
str_value = "hello"
self.assertEqual(set_msg_data("str", str_value), "hello")
# 测试默认转换
default_value = 123
self.assertEqual(set_msg_data("unknown_type", default_value), "123")
if __name__ == "__main__":
unittest.main()

47
unilabos/test/ros/msgs/test_runner.py
View File

@@ -1,47 +0,0 @@
"""
测试运行器
运行所有消息转换器的测试用例。
"""
import unittest
import sys
import os
# 添加项目根目录到路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
# 导入测试模块
from test.ros.msgs.test_basic import TestBasicFunctionality
from test.ros.msgs.test_conversion import TestBasicConversion, TestMappingConversion
from test.ros.msgs.test_mapping import TestTypeMapping, TestFieldMapping
def run_tests():
"""运行所有测试"""
# 创建测试加载器
loader = unittest.TestLoader()
# 创建测试套件
suite = unittest.TestSuite()
# 添加测试类
suite.addTests(loader.loadTestsFromTestCase(TestBasicFunctionality))
suite.addTests(loader.loadTestsFromTestCase(TestBasicConversion))
suite.addTests(loader.loadTestsFromTestCase(TestMappingConversion))
suite.addTests(loader.loadTestsFromTestCase(TestTypeMapping))
suite.addTests(loader.loadTestsFromTestCase(TestFieldMapping))
# 创建测试运行器
runner = unittest.TextTestRunner(verbosity=2)
# 运行测试
result = runner.run(suite)
# 返回结果
return result.wasSuccessful()
if __name__ == "__main__":
success = run_tests()
sys.exit(not success)

0
unilabos/test/workflow/__init__.py
View File

186
unilabos/test/workflow/example_bio.json
View File

@@ -1,186 +0,0 @@
{
"workflow": [
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_1",
"targets": "Liquid_2",
"asp_vol": 66.0,
"dis_vol": 66.0,
"asp_flow_rate": 94.0,
"dis_flow_rate": 94.0
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_2",
"targets": "Liquid_3",
"asp_vol": 58.0,
"dis_vol": 96.0,
"asp_flow_rate": 94.0,
"dis_flow_rate": 94.0
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_4",
"targets": "Liquid_2",
"asp_vol": 85.0,
"dis_vol": 170.0,
"asp_flow_rate": 94.0,
"dis_flow_rate": 94.0
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_4",
"targets": "Liquid_2",
"asp_vol": 63.333333333333336,
"dis_vol": 170.0,
"asp_flow_rate": 94.0,
"dis_flow_rate": 94.0
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_2",
"targets": "Liquid_3",
"asp_vol": 72.0,
"dis_vol": 150.0,
"asp_flow_rate": 94.0,
"dis_flow_rate": 94.0
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_4",
"targets": "Liquid_2",
"asp_vol": 85.0,
"dis_vol": 170.0,
"asp_flow_rate": 94.0,
"dis_flow_rate": 94.0
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_4",
"targets": "Liquid_2",
"asp_vol": 63.333333333333336,
"dis_vol": 170.0,
"asp_flow_rate": 94.0,
"dis_flow_rate": 94.0
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_2",
"targets": "Liquid_3",
"asp_vol": 72.0,
"dis_vol": 150.0,
"asp_flow_rate": 94.0,
"dis_flow_rate": 94.0
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_2",
"targets": "Liquid_3",
"asp_vol": 20.0,
"dis_vol": 20.0,
"asp_flow_rate": 7.6,
"dis_flow_rate": 7.6
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_5",
"targets": "Liquid_2",
"asp_vol": 6.0,
"dis_vol": 12.0,
"asp_flow_rate": 7.6,
"dis_flow_rate": 7.6
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_5",
"targets": "Liquid_2",
"asp_vol": 10.666666666666666,
"dis_vol": 12.0,
"asp_flow_rate": 7.599999999999999,
"dis_flow_rate": 7.6
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "Liquid_2",
"targets": "Liquid_6",
"asp_vol": 12.0,
"dis_vol": 10.0,
"asp_flow_rate": 7.6,
"dis_flow_rate": 7.6
}
}
],
"reagent": {
"Liquid_6": {
"slot": 1,
"well": [
"A2"
],
"labware": "elution plate"
},
"Liquid_1": {
"slot": 2,
"well": [
"A1",
"A2",
"A4"
],
"labware": "reagent reservoir"
},
"Liquid_4": {
"slot": 2,
"well": [
"A1",
"A2",
"A4"
],
"labware": "reagent reservoir"
},
"Liquid_5": {
"slot": 2,
"well": [
"A1",
"A2",
"A4"
],
"labware": "reagent reservoir"
},
"Liquid_2": {
"slot": 4,
"well": [
"A2"
],
"labware": "TAG1 plate on Magnetic Module GEN2"
},
"Liquid_3": {
"slot": 12,
"well": [
"A1"
],
"labware": "Opentrons Fixed Trash"
}
}
}

BIN
unilabos/test/workflow/example_bio_graph.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 148 KiB

63
unilabos/test/workflow/example_prcxi.json
View File

@@ -1,63 +0,0 @@
{
"steps_info": [
{
"step_number": 1,
"action": "transfer_liquid",
"parameters": {
"source": "sample supernatant",
"target": "antibody-coated well",
"volume": 100
}
},
{
"step_number": 2,
"action": "transfer_liquid",
"parameters": {
"source": "washing buffer",
"target": "antibody-coated well",
"volume": 200
}
},
{
"step_number": 3,
"action": "transfer_liquid",
"parameters": {
"source": "washing buffer",
"target": "antibody-coated well",
"volume": 200
}
},
{
"step_number": 4,
"action": "transfer_liquid",
"parameters": {
"source": "washing buffer",
"target": "antibody-coated well",
"volume": 200
}
},
{
"step_number": 5,
"action": "transfer_liquid",
"parameters": {
"source": "TMB substrate",
"target": "antibody-coated well",
"volume": 100
}
}
],
"labware_info": [
{"reagent_name": "sample supernatant", "material_name": "96深孔板", "positions": 1},
{"reagent_name": "washing buffer", "material_name": "储液槽", "positions": 2},
{"reagent_name": "TMB substrate", "material_name": "储液槽", "positions": 3},
{"reagent_name": "antibody-coated well", "material_name": "96 细胞培养皿", "positions": 4},
{"reagent_name": "", "material_name": "300μL Tip头", "positions": 5},
{"reagent_name": "", "material_name": "300μL Tip头", "positions": 6},
{"reagent_name": "", "material_name": "300μL Tip头", "positions": 7},
{"reagent_name": "", "material_name": "300μL Tip头", "positions": 8},
{"reagent_name": "", "material_name": "300μL Tip头", "positions": 9},
{"reagent_name": "", "material_name": "300μL Tip头", "positions": 10},
{"reagent_name": "", "material_name": "300μL Tip头", "positions": 11},
{"reagent_name": "", "material_name": "300μL Tip头", "positions": 13}
]
}

BIN
unilabos/test/workflow/example_prcxi_graph.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 140 KiB

BIN
unilabos/test/workflow/example_prcxi_graph_20251022_1359.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 117 KiB

94
unilabos/test/workflow/merge_workflow.py
View File

@@ -1,94 +0,0 @@
import json
import sys
from datetime import datetime
from pathlib import Path
ROOT_DIR = Path(__file__).resolve().parents[2]
if str(ROOT_DIR) not in sys.path:
sys.path.insert(0, str(ROOT_DIR))
import pytest
from scripts.workflow import build_protocol_graph, draw_protocol_graph, draw_protocol_graph_with_ports
ROOT_DIR = Path(__file__).resolve().parents[2]
if str(ROOT_DIR) not in sys.path:
sys.path.insert(0, str(ROOT_DIR))
def _normalize_steps(data):
normalized = []
for step in data:
action = step.get("action") or step.get("operation")
if not action:
continue
raw_params = step.get("parameters") or step.get("action_args") or {}
params = dict(raw_params)
if "source" in raw_params and "sources" not in raw_params:
params["sources"] = raw_params["source"]
if "target" in raw_params and "targets" not in raw_params:
params["targets"] = raw_params["target"]
description = step.get("description") or step.get("purpose")
step_dict = {"action": action, "parameters": params}
if description:
step_dict["description"] = description
normalized.append(step_dict)
return normalized
def _normalize_labware(data):
labware = {}
for item in data:
reagent_name = item.get("reagent_name")
key = reagent_name or item.get("material_name") or item.get("name")
if not key:
continue
key = str(key)
idx = 1
original_key = key
while key in labware:
idx += 1
key = f"{original_key}_{idx}"
labware[key] = {
"slot": item.get("positions") or item.get("slot"),
"labware": item.get("material_name") or item.get("labware"),
"well": item.get("well", []),
"type": item.get("type", "reagent"),
"role": item.get("role", ""),
"name": key,
}
return labware
@pytest.mark.parametrize("protocol_name", [
"example_bio",
# "bioyond_materials_liquidhandling_1",
"example_prcxi",
])
def test_build_protocol_graph(protocol_name):
data_path = Path(__file__).with_name(f"{protocol_name}.json")
with data_path.open("r", encoding="utf-8") as fp:
d = json.load(fp)
if "workflow" in d and "reagent" in d:
protocol_steps = d["workflow"]
labware_info = d["reagent"]
elif "steps_info" in d and "labware_info" in d:
protocol_steps = _normalize_steps(d["steps_info"])
labware_info = _normalize_labware(d["labware_info"])
else:
raise ValueError("Unsupported protocol format")
graph = build_protocol_graph(
labware_info=labware_info,
protocol_steps=protocol_steps,
workstation_name="PRCXi",
)
timestamp = datetime.now().strftime("%Y%m%d_%H%M")
output_path = data_path.with_name(f"{protocol_name}_graph_{timestamp}.png")
draw_protocol_graph_with_ports(graph, str(output_path))
print(graph)

0
unilabos/test/resources/__init__.py → unilabos/workflow/__init__.py
View File

547
unilabos/workflow/common.py Normal file
View File

@@ -0,0 +1,547 @@
import re
import uuid
import networkx as nx
from networkx.drawing.nx_agraph import to_agraph
import matplotlib.pyplot as plt
from typing import Dict, List, Any, Tuple, Optional
Json = Dict[str, Any]
# ---------------- Graph ----------------
class WorkflowGraph:
"""简单的有向图实现：使用 params 单层参数；inputs 内含连线；支持 node-link 导出"""
def __init__(self):
self.nodes: Dict[str, Dict[str, Any]] = {}
self.edges: List[Dict[str, Any]] = []
def add_node(self, node_id: str, **attrs):
self.nodes[node_id] = attrs
def add_edge(self, source: str, target: str, **attrs):
# 将 source_port/target_port 映射为服务端期望的 source_handle_key/target_handle_key
source_handle_key = attrs.pop("source_port", "") or attrs.pop("source_handle_key", "")
target_handle_key = attrs.pop("target_port", "") or attrs.pop("target_handle_key", "")
edge = {
"source": source,
"target": target,
"source_node_uuid": source,
"target_node_uuid": target,
"source_handle_key": source_handle_key,
"source_handle_io": attrs.pop("source_handle_io", "source"),
"target_handle_key": target_handle_key,
"target_handle_io": attrs.pop("target_handle_io", "target"),
**attrs,
}
self.edges.append(edge)
def _materialize_wiring_into_inputs(
self,
obj: Any,
inputs: Dict[str, Any],
variable_sources: Dict[str, Dict[str, Any]],
target_node_id: str,
base_path: List[str],
):
has_var = False
def walk(node: Any, path: List[str]):
nonlocal has_var
if isinstance(node, dict):
if "__var__" in node:
has_var = True
varname = node["__var__"]
placeholder = f"${{{varname}}}"
src = variable_sources.get(varname)
if src:
key = ".".join(path) # e.g. "params.foo.bar.0"
inputs[key] = {"node": src["node_id"], "output": src.get("output_name", "result")}
self.add_edge(
str(src["node_id"]),
target_node_id,
source_handle_io=src.get("output_name", "result"),
target_handle_io=key,
)
return placeholder
return {k: walk(v, path + [k]) for k, v in node.items()}
if isinstance(node, list):
return [walk(v, path + [str(i)]) for i, v in enumerate(node)]
return node
replaced = walk(obj, base_path[:])
return replaced, has_var
def add_workflow_node(
self,
node_id: int,
*,
device_key: Optional[str] = None, # 实例名，如 "ser"
resource_name: Optional[str] = None, # registry key（原 device_class）
module: Optional[str] = None,
template_name: Optional[str] = None, # 动作/模板名（原 action_key）
params: Dict[str, Any],
variable_sources: Dict[str, Dict[str, Any]],
add_ready_if_no_vars: bool = True,
prev_node_id: Optional[int] = None,
**extra_attrs,
) -> None:
"""添加工作流节点：params 单层；自动变量连线与 ready 串联；支持附加属性"""
node_id_str = str(node_id)
inputs: Dict[str, Any] = {}
params, has_var = self._materialize_wiring_into_inputs(
params, inputs, variable_sources, node_id_str, base_path=["params"]
)
if add_ready_if_no_vars and not has_var:
last_id = str(prev_node_id) if prev_node_id is not None else "-1"
inputs["ready"] = {"node": int(last_id), "output": "ready"}
self.add_edge(last_id, node_id_str, source_handle_io="ready", target_handle_io="ready")
node_obj = {
"device_key": device_key,
"resource_name": resource_name, # ✅ 新名字
"module": module,
"template_name": template_name, # ✅ 新名字
"params": params,
"inputs": inputs,
}
node_obj.update(extra_attrs or {})
self.add_node(node_id_str, parameters=node_obj)
# 顺序工作流导出（连线在 inputs，不返回 edges）
def to_dict(self) -> List[Dict[str, Any]]:
result = []
for node_id, attrs in self.nodes.items():
node = {"uuid": node_id}
params = dict(attrs.get("parameters", {}) or {})
flat = {k: v for k, v in attrs.items() if k != "parameters"}
flat.update(params)
node.update(flat)
result.append(node)
return sorted(result, key=lambda n: int(n["uuid"]) if str(n["uuid"]).isdigit() else n["uuid"])
# node-link 导出（含 edges）
def to_node_link_dict(self) -> Dict[str, Any]:
nodes_list = []
for node_id, attrs in self.nodes.items():
node_attrs = attrs.copy()
params = node_attrs.pop("parameters", {}) or {}
node_attrs.update(params)
nodes_list.append({"uuid": node_id, **node_attrs})
return {
"directed": True,
"multigraph": False,
"graph": {},
"nodes": nodes_list,
"edges": self.edges,
"links": self.edges,
}
def refactor_data(
data: List[Dict[str, Any]],
action_resource_mapping: Optional[Dict[str, str]] = None,
) -> List[Dict[str, Any]]:
"""统一的数据重构函数，根据操作类型自动选择模板
Args:
data: 原始步骤数据列表
action_resource_mapping: action 到 resource_name 的映射字典，可选
"""
refactored_data = []
# 定义操作映射，包含生物实验和有机化学的所有操作
OPERATION_MAPPING = {
# 生物实验操作
"transfer_liquid": "transfer_liquid",
"transfer": "transfer",
"incubation": "incubation",
"move_labware": "move_labware",
"oscillation": "oscillation",
# 有机化学操作
"HeatChillToTemp": "HeatChillProtocol",
"StopHeatChill": "HeatChillStopProtocol",
"StartHeatChill": "HeatChillStartProtocol",
"HeatChill": "HeatChillProtocol",
"Dissolve": "DissolveProtocol",
"Transfer": "TransferProtocol",
"Evaporate": "EvaporateProtocol",
"Recrystallize": "RecrystallizeProtocol",
"Filter": "FilterProtocol",
"Dry": "DryProtocol",
"Add": "AddProtocol",
}
UNSUPPORTED_OPERATIONS = ["Purge", "Wait", "Stir", "ResetHandling"]
for step in data:
operation = step.get("action")
if not operation or operation in UNSUPPORTED_OPERATIONS:
continue
# 处理重复操作
if operation == "Repeat":
times = step.get("times", step.get("parameters", {}).get("times", 1))
sub_steps = step.get("steps", step.get("parameters", {}).get("steps", []))
for i in range(int(times)):
sub_data = refactor_data(sub_steps, action_resource_mapping)
refactored_data.extend(sub_data)
continue
# 获取模板名称
template_name = OPERATION_MAPPING.get(operation)
if not template_name:
# 自动推断模板类型
if operation.lower() in ["transfer", "incubation", "move_labware", "oscillation"]:
template_name = f"biomek-{operation}"
else:
template_name = f"{operation}Protocol"
# 获取 resource_name
resource_name = f"device.{operation.lower()}"
if action_resource_mapping:
resource_name = action_resource_mapping.get(operation, resource_name)
# 获取步骤编号，生成 name 字段
step_number = step.get("step_number")
name = f"Step {step_number}" if step_number is not None else None
# 创建步骤数据
step_data = {
"template_name": template_name,
"resource_name": resource_name,
"description": step.get("description", step.get("purpose", f"{operation} operation")),
"lab_node_type": "Device",
"param": step.get("parameters", step.get("action_args", {})),
"footer": f"{template_name}-{resource_name}",
}
if name:
step_data["name"] = name
refactored_data.append(step_data)
return refactored_data
def build_protocol_graph(
labware_info: List[Dict[str, Any]],
protocol_steps: List[Dict[str, Any]],
workstation_name: str,
action_resource_mapping: Optional[Dict[str, str]] = None,
) -> WorkflowGraph:
"""统一的协议图构建函数，根据设备类型自动选择构建逻辑
Args:
labware_info: labware 信息字典
protocol_steps: 协议步骤列表
workstation_name: 工作站名称
action_resource_mapping: action 到 resource_name 的映射字典，可选
"""
G = WorkflowGraph()
resource_last_writer = {}
protocol_steps = refactor_data(protocol_steps, action_resource_mapping)
# 有机化学&移液站协议图构建
WORKSTATION_ID = workstation_name
# 为所有labware创建资源节点
res_index = 0
for labware_id, item in labware_info.items():
# item_id = item.get("id") or item.get("name", f"item_{uuid.uuid4()}")
node_id = str(uuid.uuid4())
# 判断节点类型
if "Rack" in str(labware_id) or "Tip" in str(labware_id):
lab_node_type = "Labware"
description = f"Prepare Labware: {labware_id}"
liquid_type = []
liquid_volume = []
elif item.get("type") == "hardware" or "reactor" in str(labware_id).lower():
if "reactor" not in str(labware_id).lower():
continue
lab_node_type = "Sample"
description = f"Prepare Reactor: {labware_id}"
liquid_type = []
liquid_volume = []
else:
lab_node_type = "Reagent"
description = f"Add Reagent to Flask: {labware_id}"
liquid_type = [labware_id]
liquid_volume = [1e5]
res_index += 1
G.add_node(
node_id,
template_name="create_resource",
resource_name="host_node",
name=f"Res {res_index}",
description=description,
lab_node_type=lab_node_type,
footer="create_resource-host_node",
param={
"res_id": labware_id,
"device_id": WORKSTATION_ID,
"class_name": "container",
"parent": WORKSTATION_ID,
"bind_locations": {"x": 0.0, "y": 0.0, "z": 0.0},
"liquid_input_slot": [-1],
"liquid_type": liquid_type,
"liquid_volume": liquid_volume,
"slot_on_deck": "",
},
)
resource_last_writer[labware_id] = f"{node_id}:labware"
last_control_node_id = None
# 处理协议步骤
for step in protocol_steps:
node_id = str(uuid.uuid4())
G.add_node(node_id, **step)
# 控制流
if last_control_node_id is not None:
G.add_edge(last_control_node_id, node_id, source_port="ready", target_port="ready")
last_control_node_id = node_id
# 物料流
params = step.get("param", {})
input_resources_possible_names = [
"vessel",
"to_vessel",
"from_vessel",
"reagent",
"solvent",
"compound",
"sources",
"targets",
]
for target_port in input_resources_possible_names:
resource_name = params.get(target_port)
if resource_name and resource_name in resource_last_writer:
source_node, source_port = resource_last_writer[resource_name].split(":")
G.add_edge(source_node, node_id, source_port=source_port, target_port=target_port)
output_resources = {
"vessel_out": params.get("vessel"),
"from_vessel_out": params.get("from_vessel"),
"to_vessel_out": params.get("to_vessel"),
"filtrate_out": params.get("filtrate_vessel"),
"reagent": params.get("reagent"),
"solvent": params.get("solvent"),
"compound": params.get("compound"),
"sources_out": params.get("sources"),
"targets_out": params.get("targets"),
}
for source_port, resource_name in output_resources.items():
if resource_name:
resource_last_writer[resource_name] = f"{node_id}:{source_port}"
return G
def draw_protocol_graph(protocol_graph: WorkflowGraph, output_path: str):
"""
(辅助功能) 使用 networkx 和 matplotlib 绘制协议工作流图，用于可视化。
"""
if not protocol_graph:
print("Cannot draw graph: Graph object is empty.")
return
G = nx.DiGraph()
for node_id, attrs in protocol_graph.nodes.items():
label = attrs.get("description", attrs.get("template_name", node_id[:8]))
G.add_node(node_id, label=label, **attrs)
for edge in protocol_graph.edges:
G.add_edge(edge["source"], edge["target"])
plt.figure(figsize=(20, 15))
try:
pos = nx.nx_agraph.graphviz_layout(G, prog="dot")
except Exception:
pos = nx.shell_layout(G) # Fallback layout
node_labels = {node: data["label"] for node, data in G.nodes(data=True)}
nx.draw(
G,
pos,
with_labels=False,
node_size=2500,
node_color="skyblue",
node_shape="o",
edge_color="gray",
width=1.5,
arrowsize=15,
)
nx.draw_networkx_labels(G, pos, labels=node_labels, font_size=8, font_weight="bold")
plt.title("Chemical Protocol Workflow Graph", size=15)
plt.savefig(output_path, dpi=300, bbox_inches="tight")
plt.close()
print(f" - Visualization saved to '{output_path}'")
COMPASS = {"n", "e", "s", "w", "ne", "nw", "se", "sw", "c"}
def _is_compass(port: str) -> bool:
return isinstance(port, str) and port.lower() in COMPASS
def draw_protocol_graph_with_ports(protocol_graph, output_path: str, rankdir: str = "LR"):
"""
使用 Graphviz 端口语法绘制协议工作流图。
- 若边上的 source_port/target_port 是 compass（n/e/s/w/...），直接用 compass。
- 否则自动为节点创建 record 形状并定义命名端口 <portname>。
最终由 PyGraphviz 渲染并输出到 output_path（后缀决定格式，如 .png/.svg/.pdf）。
"""
if not protocol_graph:
print("Cannot draw graph: Graph object is empty.")
return
# 1) 先用 networkx 搭建有向图，保留端口属性
G = nx.DiGraph()
for node_id, attrs in protocol_graph.nodes.items():
label = attrs.get("description", attrs.get("template_name", node_id[:8]))
# 保留一个干净的“中心标签”，用于放在 record 的中间槽
G.add_node(node_id, _core_label=str(label), **{k: v for k, v in attrs.items() if k not in ("label",)})
edges_data = []
in_ports_by_node = {} # 收集命名输入端口
out_ports_by_node = {} # 收集命名输出端口
for edge in protocol_graph.edges:
u = edge["source"]
v = edge["target"]
sp = edge.get("source_handle_key") or edge.get("source_port")
tp = edge.get("target_handle_key") or edge.get("target_port")
# 记录到图里（保留原始端口信息）
G.add_edge(u, v, source_handle_key=sp, target_handle_key=tp)
edges_data.append((u, v, sp, tp))
# 如果不是 compass，就按“命名端口”先归类，等会儿给节点造 record
if sp and not _is_compass(sp):
out_ports_by_node.setdefault(u, set()).add(str(sp))
if tp and not _is_compass(tp):
in_ports_by_node.setdefault(v, set()).add(str(tp))
# 2) 转为 AGraph，使用 Graphviz 渲染
A = to_agraph(G)
A.graph_attr.update(rankdir=rankdir, splines="true", concentrate="false", fontsize="10")
A.node_attr.update(
shape="box", style="rounded,filled", fillcolor="lightyellow", color="#999999", fontname="Helvetica"
)
A.edge_attr.update(arrowsize="0.8", color="#666666")
# 3) 为需要命名端口的节点设置 record 形状与 label
# 左列 = 输入端口；中间 = 核心标签；右列 = 输出端口
for n in A.nodes():
node = A.get_node(n)
core = G.nodes[n].get("_core_label", n)
in_ports = sorted(in_ports_by_node.get(n, []))
out_ports = sorted(out_ports_by_node.get(n, []))
# 如果该节点涉及命名端口，则用 record；否则保留原 box
if in_ports or out_ports:
def port_fields(ports):
if not ports:
return " " # 必须留一个空槽占位
# 每个端口一个小格子，<p> name
return "|".join(f"<{re.sub(r'[^A-Za-z0-9_:.|-]', '_', p)}> {p}" for p in ports)
left = port_fields(in_ports)
right = port_fields(out_ports)
# 三栏：左(入) | 中(节点名) | 右(出)
record_label = f"{{ {left} | {core} | {right} }}"
node.attr.update(shape="record", label=record_label)
else:
# 没有命名端口：普通盒子，显示核心标签
node.attr.update(label=str(core))
# 4) 给边设置 headport / tailport
# - 若端口为 compass：直接用 compass（e.g., headport="e"）
# - 若端口为命名端口：使用在 record 中定义的 <port> 名（同名即可）
for u, v, sp, tp in edges_data:
e = A.get_edge(u, v)
# Graphviz 属性：tail 是源，head 是目标
if sp:
if _is_compass(sp):
e.attr["tailport"] = sp.lower()
else:
# 与 record label 中 <port> 名一致；特殊字符已在 label 中做了清洗
e.attr["tailport"] = re.sub(r"[^A-Za-z0-9_:.|-]", "_", str(sp))
if tp:
if _is_compass(tp):
e.attr["headport"] = tp.lower()
else:
e.attr["headport"] = re.sub(r"[^A-Za-z0-9_:.|-]", "_", str(tp))
# 可选：若想让边更贴边缘，可设置 constraint/spline 等
# e.attr["arrowhead"] = "vee"
# 5) 输出
A.draw(output_path, prog="dot")
print(f" - Port-aware workflow rendered to '{output_path}'")
# ---------------- Registry Adapter ----------------
class RegistryAdapter:
"""根据 module 的类名（冒号右侧）反查 registry 的 resource_name（原 device_class），并抽取参数顺序"""
def __init__(self, device_registry: Dict[str, Any]):
self.device_registry = device_registry or {}
self.module_class_to_resource = self._build_module_class_index()
def _build_module_class_index(self) -> Dict[str, str]:
idx = {}
for resource_name, info in self.device_registry.items():
module = info.get("module")
if isinstance(module, str) and ":" in module:
cls = module.split(":")[-1]
idx[cls] = resource_name
idx[cls.lower()] = resource_name
return idx
def resolve_resource_by_classname(self, class_name: str) -> Optional[str]:
if not class_name:
return None
return self.module_class_to_resource.get(class_name) or self.module_class_to_resource.get(class_name.lower())
def get_device_module(self, resource_name: Optional[str]) -> Optional[str]:
if not resource_name:
return None
return self.device_registry.get(resource_name, {}).get("module")
def get_actions(self, resource_name: Optional[str]) -> Dict[str, Any]:
if not resource_name:
return {}
return (self.device_registry.get(resource_name, {}).get("class", {}).get("action_value_mappings", {})) or {}
def get_action_schema(self, resource_name: Optional[str], template_name: str) -> Optional[Json]:
return (self.get_actions(resource_name).get(template_name) or {}).get("schema")
def get_action_goal_default(self, resource_name: Optional[str], template_name: str) -> Json:
return (self.get_actions(resource_name).get(template_name) or {}).get("goal_default", {}) or {}
def get_action_input_keys(self, resource_name: Optional[str], template_name: str) -> List[str]:
schema = self.get_action_schema(resource_name, template_name) or {}
goal = (schema.get("properties") or {}).get("goal") or {}
props = goal.get("properties") or {}
required = goal.get("required") or []
return list(dict.fromkeys(required + list(props.keys())))

356
unilabos/workflow/convert_from_json.py Normal file
View File

@@ -0,0 +1,356 @@
"""
JSON 工作流转换模块
提供从多种 JSON 格式转换为统一工作流格式的功能。
支持的格式：
1. workflow/reagent 格式
2. steps_info/labware_info 格式
"""
import json
from os import PathLike
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple, Union
from unilabos.workflow.common import WorkflowGraph, build_protocol_graph
from unilabos.registry.registry import lab_registry
def get_action_handles(resource_name: str, template_name: str) -> Dict[str, List[str]]:
"""
从 registry 获取指定设备和动作的 handles 配置
Args:
resource_name: 设备资源名称，如 "liquid_handler.prcxi"
template_name: 动作模板名称，如 "transfer_liquid"
Returns:
包含 source 和 target handler_keys 的字典:
{"source": ["sources_out", "targets_out", ...], "target": ["sources", "targets", ...]}
"""
result = {"source": [], "target": []}
device_info = lab_registry.device_type_registry.get(resource_name, {})
if not device_info:
return result
action_mappings = device_info.get("class", {}).get("action_value_mappings", {})
action_config = action_mappings.get(template_name, {})
handles = action_config.get("handles", {})
if isinstance(handles, dict):
# 处理 input handles (作为 target)
for handle in handles.get("input", []):
handler_key = handle.get("handler_key", "")
if handler_key:
result["source"].append(handler_key)
# 处理 output handles (作为 source)
for handle in handles.get("output", []):
handler_key = handle.get("handler_key", "")
if handler_key:
result["target"].append(handler_key)
return result
def validate_workflow_handles(graph: WorkflowGraph) -> Tuple[bool, List[str]]:
"""
校验工作流图中所有边的句柄配置是否正确
Args:
graph: 工作流图对象
Returns:
(is_valid, errors): 是否有效，错误信息列表
"""
errors = []
nodes = graph.nodes
for edge in graph.edges:
left_uuid = edge.get("source")
right_uuid = edge.get("target")
# target_handle_key是target, right的输入节点（入节点）
# source_handle_key是source, left的输出节点（出节点）
right_source_conn_key = edge.get("target_handle_key", "")
left_target_conn_key = edge.get("source_handle_key", "")
# 获取源节点和目标节点信息
left_node = nodes.get(left_uuid, {})
right_node = nodes.get(right_uuid, {})
left_res_name = left_node.get("resource_name", "")
left_template_name = left_node.get("template_name", "")
right_res_name = right_node.get("resource_name", "")
right_template_name = right_node.get("template_name", "")
# 获取源节点的 output handles
left_node_handles = get_action_handles(left_res_name, left_template_name)
target_valid_keys = left_node_handles.get("target", [])
target_valid_keys.append("ready")
# 获取目标节点的 input handles
right_node_handles = get_action_handles(right_res_name, right_template_name)
source_valid_keys = right_node_handles.get("source", [])
source_valid_keys.append("ready")
# 如果节点配置了 output handles，则 source_port 必须有效
if not right_source_conn_key:
node_name = left_node.get("name", left_uuid[:8])
errors.append(f"源节点 '{node_name}' 的 source_handle_key 为空，" f"应设置为: {source_valid_keys}")
elif right_source_conn_key not in source_valid_keys:
node_name = left_node.get("name", left_uuid[:8])
errors.append(
f"源节点 '{node_name}' 的 source 端点 '{right_source_conn_key}' 不存在，" f"支持的端点: {source_valid_keys}"
)
# 如果节点配置了 input handles，则 target_port 必须有效
if not left_target_conn_key:
node_name = right_node.get("name", right_uuid[:8])
errors.append(f"目标节点 '{node_name}' 的 target_handle_key 为空，" f"应设置为: {target_valid_keys}")
elif left_target_conn_key not in target_valid_keys:
node_name = right_node.get("name", right_uuid[:8])
errors.append(
f"目标节点 '{node_name}' 的 target 端点 '{left_target_conn_key}' 不存在，"
f"支持的端点: {target_valid_keys}"
)
return len(errors) == 0, errors
# action 到 resource_name 的映射
ACTION_RESOURCE_MAPPING: Dict[str, str] = {
# 生物实验操作
"transfer_liquid": "liquid_handler.prcxi",
"transfer": "liquid_handler.prcxi",
"incubation": "incubator.prcxi",
"move_labware": "labware_mover.prcxi",
"oscillation": "shaker.prcxi",
# 有机化学操作
"HeatChillToTemp": "heatchill.chemputer",
"StopHeatChill": "heatchill.chemputer",
"StartHeatChill": "heatchill.chemputer",
"HeatChill": "heatchill.chemputer",
"Dissolve": "stirrer.chemputer",
"Transfer": "liquid_handler.chemputer",
"Evaporate": "rotavap.chemputer",
"Recrystallize": "reactor.chemputer",
"Filter": "filter.chemputer",
"Dry": "dryer.chemputer",
"Add": "liquid_handler.chemputer",
}
def normalize_steps(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
将不同格式的步骤数据规范化为统一格式
支持的输入格式：
- action + parameters
- action + action_args
- operation + parameters
Args:
data: 原始步骤数据列表
Returns:
规范化后的步骤列表，格式为 [{"action": str, "parameters": dict, "description": str?, "step_number": int?}, ...]
"""
normalized = []
for idx, step in enumerate(data):
# 获取动作名称（支持 action 或 operation 字段）
action = step.get("action") or step.get("operation")
if not action:
continue
# 获取参数（支持 parameters 或 action_args 字段）
raw_params = step.get("parameters") or step.get("action_args") or {}
params = dict(raw_params)
# 规范化 source/target -> sources/targets
if "source" in raw_params and "sources" not in raw_params:
params["sources"] = raw_params["source"]
if "target" in raw_params and "targets" not in raw_params:
params["targets"] = raw_params["target"]
# 获取描述（支持 description 或 purpose 字段）
description = step.get("description") or step.get("purpose")
# 获取步骤编号（优先使用原始数据中的 step_number，否则使用索引+1）
step_number = step.get("step_number", idx + 1)
step_dict = {"action": action, "parameters": params, "step_number": step_number}
if description:
step_dict["description"] = description
normalized.append(step_dict)
return normalized
def normalize_labware(data: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
"""
将不同格式的 labware 数据规范化为统一的字典格式
支持的输入格式：
- reagent_name + material_name + positions
- name + labware + slot
Args:
data: 原始 labware 数据列表
Returns:
规范化后的 labware 字典，格式为 {name: {"slot": int, "labware": str, "well": list, "type": str, "role": str, "name": str}, ...}
"""
labware = {}
for item in data:
# 获取 key 名称（优先使用 reagent_name，其次是 material_name 或 name）
reagent_name = item.get("reagent_name")
key = reagent_name or item.get("material_name") or item.get("name")
if not key:
continue
key = str(key)
# 处理重复 key，自动添加后缀
idx = 1
original_key = key
while key in labware:
idx += 1
key = f"{original_key}_{idx}"
labware[key] = {
"slot": item.get("positions") or item.get("slot"),
"labware": item.get("material_name") or item.get("labware"),
"well": item.get("well", []),
"type": item.get("type", "reagent"),
"role": item.get("role", ""),
"name": key,
}
return labware
def convert_from_json(
data: Union[str, PathLike, Dict[str, Any]],
workstation_name: str = "PRCXi",
validate: bool = True,
) -> WorkflowGraph:
"""
从 JSON 数据或文件转换为 WorkflowGraph
支持的 JSON 格式：
1. {"workflow": [...], "reagent": {...}} - 直接格式
2. {"steps_info": [...], "labware_info": [...]} - 需要规范化的格式
Args:
data: JSON 文件路径、字典数据、或 JSON 字符串
workstation_name: 工作站名称，默认 "PRCXi"
validate: 是否校验句柄配置，默认 True
Returns:
WorkflowGraph: 构建好的工作流图
Raises:
ValueError: 不支持的 JSON 格式 或 句柄校验失败
FileNotFoundError: 文件不存在
json.JSONDecodeError: JSON 解析失败
"""
# 处理输入数据
if isinstance(data, (str, PathLike)):
path = Path(data)
if path.exists():
with path.open("r", encoding="utf-8") as fp:
json_data = json.load(fp)
elif isinstance(data, str):
# 尝试作为 JSON 字符串解析
json_data = json.loads(data)
else:
raise FileNotFoundError(f"文件不存在: {data}")
elif isinstance(data, dict):
json_data = data
else:
raise TypeError(f"不支持的数据类型: {type(data)}")
# 根据格式解析数据
if "workflow" in json_data and "reagent" in json_data:
# 格式1: workflow/reagent（已经是规范格式）
protocol_steps = json_data["workflow"]
labware_info = json_data["reagent"]
elif "steps_info" in json_data and "labware_info" in json_data:
# 格式2: steps_info/labware_info（需要规范化）
protocol_steps = normalize_steps(json_data["steps_info"])
labware_info = normalize_labware(json_data["labware_info"])
elif "steps" in json_data and "labware" in json_data:
# 格式3: steps/labware（另一种常见格式）
protocol_steps = normalize_steps(json_data["steps"])
if isinstance(json_data["labware"], list):
labware_info = normalize_labware(json_data["labware"])
else:
labware_info = json_data["labware"]
else:
raise ValueError(
"不支持的 JSON 格式。支持的格式：\n"
"1. {'workflow': [...], 'reagent': {...}}\n"
"2. {'steps_info': [...], 'labware_info': [...]}\n"
"3. {'steps': [...], 'labware': [...]}"
)
# 构建工作流图
graph = build_protocol_graph(
labware_info=labware_info,
protocol_steps=protocol_steps,
workstation_name=workstation_name,
action_resource_mapping=ACTION_RESOURCE_MAPPING,
)
# 校验句柄配置
if validate:
is_valid, errors = validate_workflow_handles(graph)
if not is_valid:
import warnings
for error in errors:
warnings.warn(f"句柄校验警告: {error}")
return graph
def convert_json_to_node_link(
data: Union[str, PathLike, Dict[str, Any]],
workstation_name: str = "PRCXi",
) -> Dict[str, Any]:
"""
将 JSON 数据转换为 node-link 格式的字典
Args:
data: JSON 文件路径、字典数据、或 JSON 字符串
workstation_name: 工作站名称，默认 "PRCXi"
Returns:
Dict: node-link 格式的工作流数据
"""
graph = convert_from_json(data, workstation_name)
return graph.to_node_link_dict()
def convert_json_to_workflow_list(
data: Union[str, PathLike, Dict[str, Any]],
workstation_name: str = "PRCXi",
) -> List[Dict[str, Any]]:
"""
将 JSON 数据转换为工作流列表格式
Args:
data: JSON 文件路径、字典数据、或 JSON 字符串
workstation_name: 工作站名称，默认 "PRCXi"
Returns:
List: 工作流节点列表
"""
graph = convert_from_json(data, workstation_name)
return graph.to_dict()
# 为了向后兼容，保留下划线前缀的别名
_normalize_steps = normalize_steps
_normalize_labware = normalize_labware

241
unilabos/workflow/from_python_script.py Normal file
View File

@@ -0,0 +1,241 @@
import ast
import json
from typing import Dict, List, Any, Tuple, Optional
from .common import WorkflowGraph, RegistryAdapter
Json = Dict[str, Any]
# ---------------- Converter ----------------
class DeviceMethodConverter:
"""
- 字段统一：resource_name（原 device_class）、template_name（原 action_key）
- params 单层；inputs 使用 'params.' 前缀
- SimpleGraph.add_workflow_node 负责变量连线与边
"""
def __init__(self, device_registry: Optional[Dict[str, Any]] = None):
self.graph = WorkflowGraph()
self.variable_sources: Dict[str, Dict[str, Any]] = {} # var -> {node_id, output_name}
self.instance_to_resource: Dict[str, Optional[str]] = {} # 实例名 -> resource_name
self.node_id_counter: int = 0
self.registry = RegistryAdapter(device_registry or {})
# ---- helpers ----
def _new_node_id(self) -> int:
nid = self.node_id_counter
self.node_id_counter += 1
return nid
def _assign_targets(self, targets) -> List[str]:
names: List[str] = []
import ast
if isinstance(targets, ast.Tuple):
for elt in targets.elts:
if isinstance(elt, ast.Name):
names.append(elt.id)
elif isinstance(targets, ast.Name):
names.append(targets.id)
return names
def _extract_device_instantiation(self, node) -> Optional[Tuple[str, str]]:
import ast
if not isinstance(node.value, ast.Call):
return None
callee = node.value.func
if isinstance(callee, ast.Name):
class_name = callee.id
elif isinstance(callee, ast.Attribute) and isinstance(callee.value, ast.Name):
class_name = callee.attr
else:
return None
if isinstance(node.targets[0], ast.Name):
instance = node.targets[0].id
return instance, class_name
return None
def _extract_call(self, call) -> Tuple[str, str, Dict[str, Any], str]:
import ast
owner_name, method_name, call_kind = "", "", "func"
if isinstance(call.func, ast.Attribute):
method_name = call.func.attr
if isinstance(call.func.value, ast.Name):
owner_name = call.func.value.id
call_kind = "instance" if owner_name in self.instance_to_resource else "class_or_module"
elif isinstance(call.func.value, ast.Attribute) and isinstance(call.func.value.value, ast.Name):
owner_name = call.func.value.attr
call_kind = "class_or_module"
elif isinstance(call.func, ast.Name):
method_name = call.func.id
call_kind = "func"
def pack(node):
if isinstance(node, ast.Name):
return {"type": "variable", "value": node.id}
if isinstance(node, ast.Constant):
return {"type": "constant", "value": node.value}
if isinstance(node, ast.Dict):
return {"type": "dict", "value": self._parse_dict(node)}
if isinstance(node, ast.List):
return {"type": "list", "value": self._parse_list(node)}
return {"type": "raw", "value": ast.unparse(node) if hasattr(ast, "unparse") else str(node)}
args: Dict[str, Any] = {}
pos: List[Any] = []
for a in call.args:
pos.append(pack(a))
for kw in call.keywords:
args[kw.arg] = pack(kw.value)
if pos:
args["_positional"] = pos
return owner_name, method_name, args, call_kind
def _parse_dict(self, node) -> Dict[str, Any]:
import ast
out: Dict[str, Any] = {}
for k, v in zip(node.keys, node.values):
if isinstance(k, ast.Constant):
key = str(k.value)
if isinstance(v, ast.Name):
out[key] = f"var:{v.id}"
elif isinstance(v, ast.Constant):
out[key] = v.value
elif isinstance(v, ast.Dict):
out[key] = self._parse_dict(v)
elif isinstance(v, ast.List):
out[key] = self._parse_list(v)
return out
def _parse_list(self, node) -> List[Any]:
import ast
out: List[Any] = []
for elt in node.elts:
if isinstance(elt, ast.Name):
out.append(f"var:{elt.id}")
elif isinstance(elt, ast.Constant):
out.append(elt.value)
elif isinstance(elt, ast.Dict):
out.append(self._parse_dict(elt))
elif isinstance(elt, ast.List):
out.append(self._parse_list(elt))
return out
def _normalize_var_tokens(self, x: Any) -> Any:
if isinstance(x, str) and x.startswith("var:"):
return {"__var__": x[4:]}
if isinstance(x, list):
return [self._normalize_var_tokens(i) for i in x]
if isinstance(x, dict):
return {k: self._normalize_var_tokens(v) for k, v in x.items()}
return x
def _make_params_payload(self, resource_name: Optional[str], template_name: str, call_args: Dict[str, Any]) -> Dict[str, Any]:
input_keys = self.registry.get_action_input_keys(resource_name, template_name) if resource_name else []
defaults = self.registry.get_action_goal_default(resource_name, template_name) if resource_name else {}
params: Dict[str, Any] = dict(defaults)
def unpack(p):
t, v = p.get("type"), p.get("value")
if t == "variable":
return {"__var__": v}
if t == "dict":
return self._normalize_var_tokens(v)
if t == "list":
return self._normalize_var_tokens(v)
return v
for k, p in call_args.items():
if k == "_positional":
continue
params[k] = unpack(p)
pos = call_args.get("_positional", [])
if pos:
if input_keys:
for i, p in enumerate(pos):
if i >= len(input_keys):
break
name = input_keys[i]
if name in params:
continue
params[name] = unpack(p)
else:
for i, p in enumerate(pos):
params[f"arg_{i}"] = unpack(p)
return params
# ---- handlers ----
def _on_assign(self, stmt):
import ast
inst = self._extract_device_instantiation(stmt)
if inst:
instance, code_class = inst
resource_name = self.registry.resolve_resource_by_classname(code_class)
self.instance_to_resource[instance] = resource_name
return
if isinstance(stmt.value, ast.Call):
owner, method, call_args, kind = self._extract_call(stmt.value)
if kind == "instance":
device_key = owner
resource_name = self.instance_to_resource.get(owner)
else:
device_key = owner
resource_name = self.registry.resolve_resource_by_classname(owner)
module = self.registry.get_device_module(resource_name)
params = self._make_params_payload(resource_name, method, call_args)
nid = self._new_node_id()
self.graph.add_workflow_node(
nid,
device_key=device_key,
resource_name=resource_name, # ✅
module=module,
template_name=method, # ✅
params=params,
variable_sources=self.variable_sources,
add_ready_if_no_vars=True,
prev_node_id=(nid - 1) if nid > 0 else None,
)
out_vars = self._assign_targets(stmt.targets[0])
for var in out_vars:
self.variable_sources[var] = {"node_id": nid, "output_name": "result"}
def _on_expr(self, stmt):
import ast
if not isinstance(stmt.value, ast.Call):
return
owner, method, call_args, kind = self._extract_call(stmt.value)
if kind == "instance":
device_key = owner
resource_name = self.instance_to_resource.get(owner)
else:
device_key = owner
resource_name = self.registry.resolve_resource_by_classname(owner)
module = self.registry.get_device_module(resource_name)
params = self._make_params_payload(resource_name, method, call_args)
nid = self._new_node_id()
self.graph.add_workflow_node(
nid,
device_key=device_key,
resource_name=resource_name, # ✅
module=module,
template_name=method, # ✅
params=params,
variable_sources=self.variable_sources,
add_ready_if_no_vars=True,
prev_node_id=(nid - 1) if nid > 0 else None,
)
def convert(self, python_code: str):
tree = ast.parse(python_code)
for stmt in tree.body:
if isinstance(stmt, ast.Assign):
self._on_assign(stmt)
elif isinstance(stmt, ast.Expr):
self._on_expr(stmt)
return self

131
unilabos/workflow/from_xdl.py Normal file
View File

@@ -0,0 +1,131 @@
from typing import List, Any, Dict
import xml.etree.ElementTree as ET
def convert_to_type(val: str) -> Any:
"""将字符串值转换为适当的数据类型"""
if val == "True":
return True
if val == "False":
return False
if val == "?":
return None
if val.endswith(" g"):
return float(val.split(" ")[0])
if val.endswith("mg"):
return float(val.split("mg")[0])
elif val.endswith("mmol"):
return float(val.split("mmol")[0]) / 1000
elif val.endswith("mol"):
return float(val.split("mol")[0])
elif val.endswith("ml"):
return float(val.split("ml")[0])
elif val.endswith("RPM"):
return float(val.split("RPM")[0])
elif val.endswith(" °C"):
return float(val.split(" ")[0])
elif val.endswith(" %"):
return float(val.split(" ")[0])
return val
def flatten_xdl_procedure(procedure_elem: ET.Element) -> List[ET.Element]:
"""展平嵌套的XDL程序结构"""
flattened_operations = []
TEMP_UNSUPPORTED_PROTOCOL = ["Purge", "Wait", "Stir", "ResetHandling"]
def extract_operations(element: ET.Element):
if element.tag not in ["Prep", "Reaction", "Workup", "Purification", "Procedure"]:
if element.tag not in TEMP_UNSUPPORTED_PROTOCOL:
flattened_operations.append(element)
for child in element:
extract_operations(child)
for child in procedure_elem:
extract_operations(child)
return flattened_operations
def parse_xdl_content(xdl_content: str) -> tuple:
"""解析XDL内容"""
try:
xdl_content_cleaned = "".join(c for c in xdl_content if c.isprintable())
root = ET.fromstring(xdl_content_cleaned)
synthesis_elem = root.find("Synthesis")
if synthesis_elem is None:
return None, None, None
# 解析硬件组件
hardware_elem = synthesis_elem.find("Hardware")
hardware = []
if hardware_elem is not None:
hardware = [{"id": c.get("id"), "type": c.get("type")} for c in hardware_elem.findall("Component")]
# 解析试剂
reagents_elem = synthesis_elem.find("Reagents")
reagents = []
if reagents_elem is not None:
reagents = [{"name": r.get("name"), "role": r.get("role", "")} for r in reagents_elem.findall("Reagent")]
# 解析程序
procedure_elem = synthesis_elem.find("Procedure")
if procedure_elem is None:
return None, None, None
flattened_operations = flatten_xdl_procedure(procedure_elem)
return hardware, reagents, flattened_operations
except ET.ParseError as e:
raise ValueError(f"Invalid XDL format: {e}")
def convert_xdl_to_dict(xdl_content: str) -> Dict[str, Any]:
"""
将XDL XML格式转换为标准的字典格式
Args:
xdl_content: XDL XML内容
Returns:
转换结果，包含步骤和器材信息
"""
try:
hardware, reagents, flattened_operations = parse_xdl_content(xdl_content)
if hardware is None:
return {"error": "Failed to parse XDL content", "success": False}
# 将XDL元素转换为字典格式
steps_data = []
for elem in flattened_operations:
# 转换参数类型
parameters = {}
for key, val in elem.attrib.items():
converted_val = convert_to_type(val)
if converted_val is not None:
parameters[key] = converted_val
step_dict = {
"operation": elem.tag,
"parameters": parameters,
"description": elem.get("purpose", f"Operation: {elem.tag}"),
}
steps_data.append(step_dict)
# 合并硬件和试剂为统一的labware_info格式
labware_data = []
labware_data.extend({"id": hw["id"], "type": "hardware", **hw} for hw in hardware)
labware_data.extend({"name": reagent["name"], "type": "reagent", **reagent} for reagent in reagents)
return {
"success": True,
"steps": steps_data,
"labware": labware_data,
"message": f"Successfully converted XDL to dict format. Found {len(steps_data)} steps and {len(labware_data)} labware items.",
}
except Exception as e:
error_msg = f"XDL conversion failed: {str(e)}"
return {"error": error_msg, "success": False}

138
unilabos/workflow/wf_utils.py Normal file
View File

@@ -0,0 +1,138 @@
"""
工作流工具模块
提供工作流上传等功能
"""
import json
import os
import uuid
from typing import Any, Dict, List, Optional
from unilabos.utils.banner_print import print_status
def _is_node_link_format(data: Dict[str, Any]) -> bool:
"""检查数据是否为 node-link 格式"""
return "nodes" in data and "edges" in data
def _convert_to_node_link(workflow_file: str, workflow_data: Dict[str, Any]) -> Dict[str, Any]:
"""
将非 node-link 格式的工作流数据转换为 node-link 格式
Args:
workflow_file: 工作流文件路径（用于日志）
workflow_data: 原始工作流数据
Returns:
node-link 格式的工作流数据
"""
from unilabos.workflow.convert_from_json import convert_json_to_node_link
print_status(f"检测到非 node-link 格式，正在转换...", "info")
node_link_data = convert_json_to_node_link(workflow_data)
print_status(f"转换完成", "success")
return node_link_data
def upload_workflow(
workflow_file: str,
workflow_name: Optional[str] = None,
tags: Optional[List[str]] = None,
published: bool = False,
) -> Dict[str, Any]:
"""
上传工作流到服务器
支持的输入格式：
1. node-link 格式: {"nodes": [...], "edges": [...]}
2. workflow/reagent 格式: {"workflow": [...], "reagent": {...}}
3. steps_info/labware_info 格式: {"steps_info": [...], "labware_info": [...]}
4. steps/labware 格式: {"steps": [...], "labware": [...]}
Args:
workflow_file: 工作流文件路径（JSON格式）
workflow_name: 工作流名称，如果不提供则从文件中读取或使用文件名
tags: 工作流标签列表，默认为空列表
published: 是否发布工作流，默认为False
Returns:
Dict: API响应数据
"""
# 延迟导入，避免在配置文件加载之前初始化 http_client
from unilabos.app.web import http_client
if not os.path.exists(workflow_file):
print_status(f"工作流文件不存在: {workflow_file}", "error")
return {"code": -1, "message": f"文件不存在: {workflow_file}"}
# 读取工作流文件
try:
with open(workflow_file, "r", encoding="utf-8") as f:
workflow_data = json.load(f)
except json.JSONDecodeError as e:
print_status(f"工作流文件JSON解析失败: {e}", "error")
return {"code": -1, "message": f"JSON解析失败: {e}"}
# 自动检测并转换格式
if not _is_node_link_format(workflow_data):
try:
workflow_data = _convert_to_node_link(workflow_file, workflow_data)
except Exception as e:
print_status(f"工作流格式转换失败: {e}", "error")
return {"code": -1, "message": f"格式转换失败: {e}"}
# 提取工作流数据
nodes = workflow_data.get("nodes", [])
edges = workflow_data.get("edges", [])
workflow_uuid_val = workflow_data.get("workflow_uuid", str(uuid.uuid4()))
wf_name_from_file = workflow_data.get("workflow_name", os.path.basename(workflow_file).replace(".json", ""))
# 确定工作流名称
final_name = workflow_name or wf_name_from_file
print_status(f"正在上传工作流: {final_name}", "info")
print_status(f" - 节点数量: {len(nodes)}", "info")
print_status(f" - 边数量: {len(edges)}", "info")
print_status(f" - 标签: {tags or []}", "info")
print_status(f" - 发布状态: {published}", "info")
# 调用 http_client 上传
result = http_client.workflow_import(
name=final_name,
workflow_uuid=workflow_uuid_val,
workflow_name=final_name,
nodes=nodes,
edges=edges,
tags=tags,
published=published,
)
if result.get("code") == 0:
data = result.get("data", {})
print_status("工作流上传成功！", "success")
print_status(f" - UUID: {data.get('uuid', 'N/A')}", "info")
print_status(f" - 名称: {data.get('name', 'N/A')}", "info")
else:
print_status(f"工作流上传失败: {result.get('message', '未知错误')}", "error")
return result
def handle_workflow_upload_command(args_dict: Dict[str, Any]) -> None:
"""
处理 workflow_upload 子命令
Args:
args_dict: 命令行参数字典
"""
workflow_file = args_dict.get("workflow_file")
workflow_name = args_dict.get("workflow_name")
tags = args_dict.get("tags", [])
published = args_dict.get("published", False)
if workflow_file:
upload_workflow(workflow_file, workflow_name, tags, published)
else:
print_status("未指定工作流文件路径，请使用 -f/--workflow_file 参数", "error")