open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2025-12-17 21:11:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add workflow upload func.

Browse Source
This commit is contained in:
Xuwznln
2025-12-08 19:12:05 +08:00
parent ced961050d
commit 16ee3de086
32 changed files with 811 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
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
@@ -155,14 +156,39 @@ def parse_args():
default=False,
help="Complete registry information",
)
# label
# workflow upload subcommand
workflow_parser = subparsers.add_parser(
"workflow_upload",
aliases=["wf"],
help="Upload workflow from xdl/json/python files",
)
workflow_parser.add_argument("-t", "--labeltype", default="singlepoint", type=str,
help="QM calculation type, support 'singlepoint', 'optimize' and 'dimer' currently")
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
@@ -173,9 +199,6 @@ def main():
convert_argv_dashes_to_underscores(args)
args_dict = vars(args.parse_args())
# 显示启动横幅
print_unilab_banner(args_dict)
# 环境检查 - 检查并自动安装必需的包 (可选)
if not args_dict.get("skip_env_check", False):
from unilabos.utils.environment_check import check_environment
@@ -254,18 +277,10 @@ def main():
print_status("传入了sk参数，优先采用传入参数！", "info")
BasicConfig.working_dir = working_dir
# 显示启动横幅
print_unilab_banner(args_dict)
workflow_upload = args_dict.get("command") in ("workflow_upload", "wf")
#####################################
######## 启动设备接入端(主入口) ########
#####################################
launch(args_dict)
def launch(args_dict: Dict[str, Any]):
# 使用远程资源启动
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
@@ -301,11 +316,36 @@ def launch(args_dict: Dict[str, Any]):
from unilabos.resources.graphio import modify_to_backend_format
from unilabos.ros.nodes.resource_tracker import ResourceTreeSet, ResourceDict
# 显示启动横幅
print_unilab_banner(args_dict)
# 注册表
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)
@@ -382,20 +422,6 @@ def launch(args_dict: Dict[str, Any]):
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:
@@ -410,6 +436,7 @@ def launch(args_dict: Dict[str, Any]):
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)
@@ -451,16 +478,13 @@ def launch(args_dict: Dict[str, Any]):
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()

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

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 unilabos.workflow.common 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

183
unilabos/workflow/common.py
View File

@@ -10,6 +10,7 @@ Json = Dict[str, Any]
# ---------------- Graph ----------------
class WorkflowGraph:
"""简单的有向图实现：使用 params 单层参数；inputs 内含连线；支持 node-link 导出"""
@@ -21,20 +22,31 @@ class WorkflowGraph:
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
**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]):
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]):
@@ -48,9 +60,12 @@ class WorkflowGraph:
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)
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):
@@ -60,18 +75,20 @@ class WorkflowGraph:
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:
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] = {}
@@ -87,9 +104,9 @@ class WorkflowGraph:
node_obj = {
"device_key": device_key,
"resource_name": resource_name, # ✅ 新名字
"resource_name": resource_name, # ✅ 新名字
"module": module,
"template_name": template_name, # ✅ 新名字
"template_name": template_name, # ✅ 新名字
"params": params,
"inputs": inputs,
}
@@ -100,13 +117,13 @@ class WorkflowGraph:
def to_dict(self) -> List[Dict[str, Any]]:
result = []
for node_id, attrs in self.nodes.items():
node = {"id": node_id}
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["id"]) if str(n["id"]).isdigit() else n["id"])
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]:
@@ -115,12 +132,27 @@ class WorkflowGraph:
node_attrs = attrs.copy()
params = node_attrs.pop("parameters", {}) or {}
node_attrs.update(params)
nodes_list.append({"id": node_id, **node_attrs})
return {"directed": True, "multigraph": False, "graph": {}, "nodes": nodes_list, "edges": self.edges, "links": self.edges}
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]]) -> List[Dict[str, Any]]:
"""统一的数据重构函数，根据操作类型自动选择模板"""
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 = []
# 定义操作映射，包含生物实验和有机化学的所有操作
@@ -157,43 +189,67 @@ def refactor_data(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
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)
sub_data = refactor_data(sub_steps, action_resource_mapping)
refactored_data.extend(sub_data)
continue
# 获取模板名称
template = OPERATION_MAPPING.get(operation)
if not template:
template_name = OPERATION_MAPPING.get(operation)
if not template_name:
# 自动推断模板类型
if operation.lower() in ["transfer", "incubation", "move_labware", "oscillation"]:
template = f"biomek-{operation}"
template_name = f"biomek-{operation}"
else:
template = f"{operation}Protocol"
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": template,
"template_name": template_name,
"resource_name": resource_name,
"description": step.get("description", step.get("purpose", f"{operation} operation")),
"lab_node_type": "Device",
"parameters": step.get("parameters", step.get("action_args", {})),
"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
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)
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())
@@ -217,13 +273,16 @@ def build_protocol_graph(
liquid_type = [labware_id]
liquid_volume = [1e5]
res_index += 1
G.add_node(
node_id,
template_name=f"create_resource",
template_name="create_resource",
resource_name="host_node",
name=f"Res {res_index}",
description=description,
lab_node_type=lab_node_type,
params={
footer="create_resource-host_node",
param={
"res_id": labware_id,
"device_id": WORKSTATION_ID,
"class_name": "container",
@@ -234,7 +293,6 @@ def build_protocol_graph(
"liquid_volume": liquid_volume,
"slot_on_deck": "",
},
role=item.get("role", ""),
)
resource_last_writer[labware_id] = f"{node_id}:labware"
@@ -251,7 +309,7 @@ def build_protocol_graph(
last_control_node_id = node_id
# 物料流
params = step.get("parameters", {})
params = step.get("param", {})
input_resources_possible_names = [
"vessel",
"to_vessel",
@@ -299,7 +357,7 @@ def draw_protocol_graph(protocol_graph: WorkflowGraph, output_path: str):
G = nx.DiGraph()
for node_id, attrs in protocol_graph.nodes.items():
label = attrs.get("description", attrs.get("template", node_id[:8]))
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:
@@ -331,11 +389,13 @@ def draw_protocol_graph(protocol_graph: WorkflowGraph, output_path: str):
print(f" - Visualization saved to '{output_path}'")
COMPASS = {"n","e","s","w","ne","nw","se","sw","c"}
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 端口语法绘制协议工作流图。
@@ -350,22 +410,22 @@ def draw_protocol_graph_with_ports(protocol_graph, output_path: str, rankdir: st
# 1) 先用 networkx 搭建有向图，保留端口属性
G = nx.DiGraph()
for node_id, attrs in protocol_graph.nodes.items():
label = attrs.get("description", attrs.get("template", node_id[:8]))
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",)})
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 = {} # 收集命名输入端口
in_ports_by_node = {} # 收集命名输入端口
out_ports_by_node = {} # 收集命名输出端口
for edge in protocol_graph.edges:
u = edge["source"]
v = edge["target"]
sp = edge.get("source_port")
tp = edge.get("target_port")
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_port=sp, target_port=tp)
G.add_edge(u, v, source_handle_key=sp, target_handle_key=tp)
edges_data.append((u, v, sp, tp))
# 如果不是 compass，就按“命名端口”先归类，等会儿给节点造 record
@@ -377,7 +437,9 @@ def draw_protocol_graph_with_ports(protocol_graph, output_path: str, rankdir: st
# 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.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
@@ -386,18 +448,19 @@ def draw_protocol_graph_with_ports(protocol_graph, output_path: str, rankdir: st
node = A.get_node(n)
core = G.nodes[n].get("_core_label", n)
in_ports = sorted(in_ports_by_node.get(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)
left = port_fields(in_ports)
right = port_fields(out_ports)
# 三栏：左(入) | 中(节点名) | 右(出)
@@ -410,7 +473,7 @@ def draw_protocol_graph_with_ports(protocol_graph, output_path: str, rankdir: st
# 4) 给边设置 headport / tailport
# - 若端口为 compass：直接用 compass（e.g., headport="e"）
# - 若端口为命名端口：使用在 record 中定义的 <port> 名（同名即可）
for (u, v, sp, tp) in edges_data:
for u, v, sp, tp in edges_data:
e = A.get_edge(u, v)
# Graphviz 属性：tail 是源，head 是目标
@@ -419,13 +482,13 @@ def draw_protocol_graph_with_ports(protocol_graph, output_path: str, rankdir: st
e.attr["tailport"] = sp.lower()
else:
# 与 record label 中 <port> 名一致；特殊字符已在 label 中做了清洗
e.attr["tailport"] = re.sub(r'[^A-Za-z0-9_:.|-]', '_', str(sp))
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))
e.attr["headport"] = re.sub(r"[^A-Za-z0-9_:.|-]", "_", str(tp))
# 可选：若想让边更贴边缘，可设置 constraint/spline 等
# e.attr["arrowhead"] = "vee"
@@ -433,11 +496,14 @@ def draw_protocol_graph_with_ports(protocol_graph, output_path: str, rankdir: st
# 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()
@@ -455,8 +521,7 @@ class RegistryAdapter:
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()))
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:
@@ -466,9 +531,7 @@ class RegistryAdapter:
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 {}
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")

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

24
unilabos/workflow/from_labwares_and_steps.py
View File

@@ -1,24 +0,0 @@
import json
from os import PathLike
from unilabos.workflow.common import build_protocol_graph
def from_labwares_and_steps(data_path: PathLike):
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",
)

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")