From 8d1c7b183782b41f182b955009c0df32953d527c Mon Sep 17 00:00:00 2001 From: jgwy Date: Mon, 20 Jul 2026 01:57:11 +0800 Subject: [PATCH] feat(xdl): add optional XDL workflow bridge Provide a reusable XDL-to-workflow API with explicit station bindings so agents can prepare standard Uni-Lab workflows without changing native execution. --- docs/user_guide/xdl_bridge.md | 45 ++++++ tests/xdl_bridge/test_bridge.py | 52 +++++++ .../comprehensive_protocol/xdl_bridge.yaml | 39 +++++ unilabos/xdl_bridge/__init__.py | 63 ++++++++ unilabos/xdl_bridge/builder.py | 139 ++++++++++++++++++ unilabos/xdl_bridge/contracts.py | 65 ++++++++ unilabos/xdl_bridge/models.py | 21 +++ unilabos/xdl_bridge/parser.py | 92 ++++++++++++ unilabos/xdl_bridge/profile.py | 119 +++++++++++++++ 9 files changed, 635 insertions(+) create mode 100644 docs/user_guide/xdl_bridge.md create mode 100644 tests/xdl_bridge/test_bridge.py create mode 100644 unilabos/test/experiments/comprehensive_protocol/xdl_bridge.yaml create mode 100644 unilabos/xdl_bridge/__init__.py create mode 100644 unilabos/xdl_bridge/builder.py create mode 100644 unilabos/xdl_bridge/contracts.py create mode 100644 unilabos/xdl_bridge/models.py create mode 100644 unilabos/xdl_bridge/parser.py create mode 100644 unilabos/xdl_bridge/profile.py diff --git a/docs/user_guide/xdl_bridge.md b/docs/user_guide/xdl_bridge.md new file mode 100644 index 000000000..5e93ebacb --- /dev/null +++ b/docs/user_guide/xdl_bridge.md @@ -0,0 +1,45 @@ +# XDL Bridge + +`unilabos.xdl_bridge` 是可选模块，用于把 AI 或用户生成的 XDL 转为标准 +Uni-Lab workflow。它不会改变已有 workflow、设备调度或驱动行为。 + +## 使用流程 + +1. 用户与 Agent 确认要使用的实验室设备。 +2. Agent 生成 XDL，并选择该工站对应的 bridge profile。 +3. Agent 调用 `build_xdl_workflow()` 或 `upload_xdl_workflow()`。 +4. workflow 上传到玻尔跃迁后，用户在 Workflow 页面检查并手动启动。 + +启动 Uni-Lab edge 仍使用原有命令。例如 comprehensive 预设工站： + +```bash +unilab -g unilabos/test/experiments/comprehensive_protocol/comprehensive_station.json \ + --upload_registry \ + --addr https://leap-lab.bohrium.com/api/v1 \ + --disable_browser +``` + +AK/SK 应通过命令行、环境变量或会话注入，不能写入 XDL、profile 或日志。 + +## Python API + +```python +from unilabos.xdl_bridge import build_xdl_workflow, upload_xdl_workflow + +profile = "unilabos/test/experiments/comprehensive_protocol/xdl_bridge.yaml" +workflow = build_xdl_workflow("experiment.xdl", profile) +result = upload_xdl_workflow("experiment.xdl", profile, tags=["xdl"]) +``` + +## Profile + +Profile 只绑定目标工站： + +- 设备图与 registry； +- 工作站 ID； +- XDL 硬件角色到工站资源 ID 的映射； +- `virtual` 或 `real` 运行模式。 + +XDL 操作到 `TransferProtocol`、`HeatChillProtocol` 等 Uni-Lab Protocol 的映射是 +模块共享合同，不随工站复制。若工站缺少某个 Protocol、handle 或资源绑定，bridge 在 +上传前报错。 diff --git a/tests/xdl_bridge/test_bridge.py b/tests/xdl_bridge/test_bridge.py new file mode 100644 index 000000000..369650524 --- /dev/null +++ b/tests/xdl_bridge/test_bridge.py @@ -0,0 +1,52 @@ +from pathlib import Path + +from unilabos.xdl_bridge import build_xdl_workflow, load_station_profile + + +ROOT = Path(__file__).parents[2] +COMPREHENSIVE_PROFILE = ( + ROOT + / "unilabos" + / "test" + / "experiments" + / "comprehensive_protocol" + / "xdl_bridge.yaml" +) + + +def test_comprehensive_profile_uses_shared_protocol_contract(): + profile = load_station_profile(COMPREHENSIVE_PROFILE) + + assert profile.workstation_id == "OrganicSynthesisStation" + assert profile.operation("Transfer")["template"] == "PumpTransferProtocol" + assert profile.bind_component("reactor", "reactor") == "main_reactor" + assert profile.operation("FilterThrough")["overrides"]["filter_through"] == "filter_1" + assert profile.operation("RunColumn")["overrides"]["column"] == "column_1" + + +def test_xdl_builds_standard_unilab_workflow_for_selected_station(tmp_path): + xdl = tmp_path / "transfer.xdl" + xdl.write_text( + """ + + + + + + + + + + +""", + encoding="utf-8", + ) + + workflow = build_xdl_workflow(xdl, COMPREHENSIVE_PROFILE, name="transfer") + + node = workflow["nodes"][0] + assert node["resource_name"] == "workstation" + assert node["device_name"] == "OrganicSynthesisStation" + assert node["template_name"] == "PumpTransferProtocol" + assert node["param"]["from_vessel"] == "main_reactor" + assert node["param"]["to_vessel"] == "separator_1" diff --git a/unilabos/test/experiments/comprehensive_protocol/xdl_bridge.yaml b/unilabos/test/experiments/comprehensive_protocol/xdl_bridge.yaml new file mode 100644 index 000000000..dfabf610b --- /dev/null +++ b/unilabos/test/experiments/comprehensive_protocol/xdl_bridge.yaml @@ -0,0 +1,39 @@ +station: + graph: comprehensive_station.json + registry: ../../../registry/devices/work_station.yaml + workstation_id: OrganicSynthesisStation + resource_name: workstation + mode: virtual + +hardware: + ids: + reactor: main_reactor + separator: separator_1 + rotavap: rotavap_1 + filter: filter_1 + flask_organic: collection_bottle_1 + recryst_flask: collection_bottle_2 + distill_flask: collection_bottle_2 + flask_product: collection_bottle_3 + dropping_funnel_a: reagent_bottle_4 + dropping_funnel_b: reagent_bottle_5 + waste: waste_bottle_1 + column: column_1 + silica_gel: column_1 + types: + reactor: main_reactor + separator: separator_1 + rotavap: rotavap_1 + filter: filter_1 + flask: collection_bottle_1 + column: column_1 + +operation_overrides: + Transfer: + template: PumpTransferProtocol + FilterThrough: + overrides: + filter_through: filter_1 + RunColumn: + overrides: + column: column_1 diff --git a/unilabos/xdl_bridge/__init__.py b/unilabos/xdl_bridge/__init__.py new file mode 100644 index 000000000..0edab7280 --- /dev/null +++ b/unilabos/xdl_bridge/__init__.py @@ -0,0 +1,63 @@ +"""Optional XDL-to-Uni-Lab workflow bridge. + +The bridge translates portable XDL into the existing Uni-Lab workflow contract. +It does not start devices or alter native workflow execution. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from .builder import build_workflow, validate_workflow +from .parser import parse_xdl +from .profile import StationProfile, load_station_profile + + +def build_xdl_workflow( + xdl_path: str | Path, profile_path: str | Path, *, name: str | None = None +) -> dict[str, Any]: + procedure = parse_xdl(xdl_path) + profile = load_station_profile(profile_path) + payload = build_workflow(procedure, profile, name=name or Path(xdl_path).stem) + validate_workflow(payload, profile) + return payload + + +def upload_xdl_workflow( + xdl_path: str | Path, + profile_path: str | Path, + *, + name: str | None = None, + tags: list[str] | None = None, + description: str = "", + client: Any = None, +) -> dict[str, Any]: + payload = build_xdl_workflow(xdl_path, profile_path, name=name) + if client is None: + from unilabos.app.web import http_client as client + workflow_name = name or Path(xdl_path).stem + response = client.workflow_import( + name=workflow_name, + workflow_uuid=payload["workflow_uuid"], + workflow_name=workflow_name, + nodes=payload["nodes"], + edges=payload["edges"], + tags=tags or [], + description=description, + published=False, + ) + if response.get("code") != 0: + raise RuntimeError(f"Workflow upload failed: {response}") + return response + + +__all__ = [ + "StationProfile", + "build_xdl_workflow", + "build_workflow", + "load_station_profile", + "parse_xdl", + "upload_xdl_workflow", + "validate_workflow", +] diff --git a/unilabos/xdl_bridge/builder.py b/unilabos/xdl_bridge/builder.py new file mode 100644 index 000000000..8048e12a3 --- /dev/null +++ b/unilabos/xdl_bridge/builder.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +from copy import deepcopy +import json +import uuid +from typing import Any + +from .models import CanonicalProcedure +from .profile import StationProfile + + +class BridgeValidationError(ValueError): + pass + + +_VESSEL_KEYS = { + "vessel", + "from_vessel", + "to_vessel", + "separation_vessel", + "filtrate_vessel", + "waste_phase_to_vessel", + "product_vessel", + "waste_vessel", +} + + +def _normalize_scalar(key: str, value: Any) -> Any: + if not isinstance(value, str): + return value + if value == "true": + return True + if value == "false": + return False + if key == "repeats": + try: + return int(value) + except ValueError: + return value + return value + + +def _edge(source: str, target: str, source_handle: str, target_handle: str) -> dict[str, str]: + return { + "source": source, + "target": target, + "source_node_uuid": source, + "target_node_uuid": target, + "source_handle_key": source_handle, + "source_handle_io": "source", + "target_handle_key": target_handle, + "target_handle_io": "target", + } + + +def build_workflow( + procedure: CanonicalProcedure, profile: StationProfile, *, name: str +) -> dict[str, Any]: + bindings = { + component["id"]: profile.bind_component( + component["id"], component.get("type", "") + ) + for component in procedure.components + } + nodes: list[dict[str, Any]] = [] + edges: list[dict[str, str]] = [] + latest_output: dict[str, tuple[str, str]] = {} + previous_node: str | None = None + for step in procedure.steps: + operation = profile.operation(step.operation) + node_id = str(uuid.uuid4()) + parameters = deepcopy(step.parameters) + for old, new in operation.get("parameter_aliases", {}).items(): + if old in parameters and new not in parameters: + parameters[new] = parameters.pop(old) + for key, value in operation.get("defaults", {}).items(): + parameters.setdefault(key, value) + parameters.update(operation.get("overrides", {})) + for key, value in tuple(parameters.items()): + if key in _VESSEL_KEYS and isinstance(value, str): + try: + parameters[key] = value if value in profile.graph_nodes else bindings[value] + except KeyError as exc: + raise BridgeValidationError( + f"{step.source_path}: unbound vessel {value!r}" + ) from exc + else: + parameters[key] = _normalize_scalar(key, value) + nodes.append( + { + "uuid": node_id, + "name": f"Step {step.sequence}", + "type": "ILab", + "lab_node_type": "ILab", + "template_name": operation["template"], + "resource_name": profile.resource_name, + "device_name": profile.workstation_id, + "description": f"{step.operation} operation", + "footer": f"{operation['template']}-{profile.resource_name}", + "param": parameters, + } + ) + for parameter, handle in operation.get("inputs", {}).items(): + resource_id = parameters.get(parameter) + if isinstance(resource_id, str) and resource_id in latest_output: + source, source_handle = latest_output[resource_id] + edges.append(_edge(source, node_id, source_handle, handle)) + if previous_node is not None: + edges.append(_edge(previous_node, node_id, "ready", "ready")) + for parameter, handle in operation.get("outputs", {}).items(): + resource_id = parameters.get(parameter) + if isinstance(resource_id, str): + latest_output[resource_id] = (node_id, handle) + previous_node = node_id + return { + "workflow_uuid": str(uuid.uuid4()), + "workflow_name": name, + "directed": True, + "multigraph": False, + "graph": {}, + "nodes": nodes, + "edges": edges, + "links": edges, + } + + +def validate_workflow(payload: dict[str, Any], profile: StationProfile) -> None: + serialized = json.dumps(payload) + for value in ("PRCXI", "liquid_handler.prcxi", "[WARN:", "device."): + if value in serialized: + raise BridgeValidationError(f"Forbidden workflow value: {value}") + for node in payload.get("nodes", []): + if node.get("resource_name") != profile.resource_name: + raise BridgeValidationError("Unexpected workflow resource") + if node.get("device_name") != profile.workstation_id: + raise BridgeValidationError("Unexpected workflow device") + for key in _VESSEL_KEYS: + if key in node.get("param", {}) and node["param"][key] not in profile.graph_nodes: + raise BridgeValidationError(f"Unbound resource {node['name']}.{key}") diff --git a/unilabos/xdl_bridge/contracts.py b/unilabos/xdl_bridge/contracts.py new file mode 100644 index 000000000..d3228c748 --- /dev/null +++ b/unilabos/xdl_bridge/contracts.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from copy import deepcopy +from typing import Any + + +STANDARD_OPERATIONS: dict[str, dict[str, Any]] = { + "Add": {"template": "AddProtocol", "inputs": {"vessel": "Vessel"}, "outputs": {"vessel": "VesselOut"}}, + "AddSolid": {"template": "AddProtocol", "inputs": {"vessel": "Vessel"}, "outputs": {"vessel": "VesselOut"}}, + "Transfer": { + "template": "TransferProtocol", + "inputs": {"from_vessel": "FromVessel", "to_vessel": "ToVessel"}, + "outputs": {"from_vessel": "FromVesselOut", "to_vessel": "ToVesselOut"}, + }, + "Stir": {"template": "StirProtocol", "inputs": {"vessel": "Vessel"}, "outputs": {"vessel": "VesselOut"}}, + "EvacuateAndRefill": {"template": "EvacuateAndRefillProtocol", "inputs": {"vessel": "Vessel"}, "outputs": {"vessel": "VesselOut"}}, + "HeatChill": {"template": "HeatChillProtocol", "inputs": {"vessel": "Vessel"}, "outputs": {"vessel": "VesselOut"}}, + "HeatChillToTemp": {"template": "HeatChillProtocol", "inputs": {"vessel": "Vessel"}, "outputs": {"vessel": "VesselOut"}}, + "Separate": { + "template": "SeparateProtocol", + "inputs": {"from_vessel": "FromVessel", "to_vessel": "ToVessel"}, + "outputs": {"from_vessel": "FromVesselOut", "to_vessel": "ToVesselOut"}, + }, + "FilterThrough": { + "template": "FilterThroughProtocol", + "inputs": {"from_vessel": "FromVessel", "to_vessel": "ToVessel"}, + "outputs": {"from_vessel": "FromVesselOut", "to_vessel": "ToVesselOut"}, + "parameter_aliases": {"through": "filter_through"}, + }, + "Evaporate": {"template": "EvaporateProtocol", "inputs": {"vessel": "Vessel"}, "outputs": {"vessel": "VesselOut"}}, + "Filter": { + "template": "FilterProtocol", + "inputs": {"vessel": "Vessel", "filtrate_vessel": "FiltrateVessel"}, + "outputs": {"vessel": "VesselOut", "filtrate_vessel": "FiltrateOut"}, + }, + "WashSolid": { + "template": "WashSolidProtocol", + "inputs": {"vessel": "Vessel", "filtrate_vessel": "filtrate_vessel"}, + "outputs": {"vessel": "VesselOut", "filtrate_vessel": "filtrate_vessel_out"}, + }, + "Recrystallize": { + "template": "RecrystallizeProtocol", + "inputs": {"vessel": "Vessel"}, + "outputs": {"vessel": "VesselOut"}, + "parameter_aliases": {"solvent": "solvent1", "solvent_volume": "volume"}, + "defaults": {"ratio": "1:0", "solvent2": "ethanol"}, + }, + "Dry": {"template": "DryProtocol", "inputs": {"vessel": "Vessel"}, "outputs": {"vessel": "VesselOut"}}, + "Distill": { + "template": "EvaporateProtocol", + "inputs": {"vessel": "Vessel"}, + "outputs": {"vessel": "VesselOut"}, + "parameter_aliases": {"vapour_temp": "temp"}, + }, + "RunColumn": { + "template": "RunColumnProtocol", + "inputs": {"from_vessel": "FromVessel", "to_vessel": "ToVessel"}, + "outputs": {"from_vessel": "FromVesselOut", "to_vessel": "ToVesselOut"}, + "parameter_aliases": {"eluting_solvent": "solvent1"}, + }, +} + + +def standard_operations() -> dict[str, dict[str, Any]]: + return deepcopy(STANDARD_OPERATIONS) diff --git a/unilabos/xdl_bridge/models.py b/unilabos/xdl_bridge/models.py new file mode 100644 index 000000000..93b394d80 --- /dev/null +++ b/unilabos/xdl_bridge/models.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +@dataclass(frozen=True) +class CanonicalStep: + operation: str + parameters: dict[str, Any] + sequence: int + source_path: str + + +@dataclass(frozen=True) +class CanonicalProcedure: + components: tuple[dict[str, str], ...] + reagents: tuple[dict[str, str], ...] + steps: tuple[CanonicalStep, ...] + source: Path diff --git a/unilabos/xdl_bridge/parser.py b/unilabos/xdl_bridge/parser.py new file mode 100644 index 000000000..372ecc958 --- /dev/null +++ b/unilabos/xdl_bridge/parser.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +from pathlib import Path +import xml.etree.ElementTree as ET + +from .models import CanonicalProcedure, CanonicalStep + + +_SECTIONS = {"Procedure", "Prep", "Reaction", "Workup", "Purification"} + + +def _synthesis(root: ET.Element) -> ET.Element: + if root.tag == "Synthesis": + return root + syntheses = root.findall("Synthesis") + if len(syntheses) != 1: + raise ValueError("XDL document must contain exactly one Synthesis element") + return syntheses[0] + + +def _ensure_unique( + records: tuple[dict[str, str], ...], *, key: str, label: str +) -> None: + seen: set[str] = set() + for record in records: + value = record.get(key, "") + if not value: + raise ValueError(f"{label.capitalize()} must declare {key}") + if value in seen: + raise ValueError(f"Duplicate {label} {key}: {value}") + seen.add(value) + + +def parse_xdl(path: str | Path) -> CanonicalProcedure: + source = Path(path) + root = ET.parse(source).getroot() + synthesis = _synthesis(root) + procedure = synthesis.find("Procedure") + if procedure is None: + raise ValueError("XDL Synthesis must contain a Procedure element") + + components = tuple( + dict(component.attrib) + for component in synthesis.findall("./Hardware/Component") + ) + reagents = tuple( + dict(reagent.attrib) for reagent in synthesis.findall("./Reagents/Reagent") + ) + _ensure_unique(components, key="id", label="hardware") + _ensure_unique(reagents, key="name", label="reagent") + steps: list[CanonicalStep] = [] + + def append(element: ET.Element, source_path: str) -> None: + if element.tag in _SECTIONS: + for index, child in enumerate(element): + append(child, f"{source_path}/{child.tag}[{index}]") + return + if element.tag == "Repeat": + raw_repeats = element.attrib.get( + "repeats", element.attrib.get("times", "1") + ) + try: + repeats = int(raw_repeats) + except ValueError as exc: + raise ValueError( + f"Repeat count must be an integer: {raw_repeats}" + ) from exc + if repeats < 1: + raise ValueError("Repeat count must be at least one") + for repeat_index in range(repeats): + for child_index, child in enumerate(element): + append( + child, + f"{source_path}/repeat[{repeat_index}]/{child.tag}[{child_index}]", + ) + return + steps.append( + CanonicalStep( + operation=element.tag, + parameters=dict(element.attrib), + sequence=len(steps) + 1, + source_path=source_path, + ) + ) + + append(procedure, "/XDL/Synthesis/Procedure") + return CanonicalProcedure( + components=components, + reagents=reagents, + steps=tuple(steps), + source=source, + ) diff --git a/unilabos/xdl_bridge/profile.py b/unilabos/xdl_bridge/profile.py new file mode 100644 index 000000000..bbddd0b3f --- /dev/null +++ b/unilabos/xdl_bridge/profile.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any +import json + +import yaml + +from .contracts import standard_operations + + +class ProfileError(ValueError): + pass + + +@dataclass(frozen=True) +class StationProfile: + path: Path + graph_path: Path + registry_path: Path + workstation_id: str + resource_name: str + mode: str + hardware_ids: dict[str, str] + hardware_types: dict[str, str] + operations: dict[str, dict[str, Any]] + graph_nodes: frozenset[str] + + def bind_component(self, component_id: str, component_type: str) -> str: + node_id = self.hardware_ids.get(component_id) or self.hardware_types.get( + component_type + ) + if node_id is None: + raise ProfileError( + f"No station binding for component {component_id!r} ({component_type!r})" + ) + if node_id not in self.graph_nodes: + raise ProfileError(f"Bound station node does not exist: {node_id}") + return node_id + + def operation(self, name: str) -> dict[str, Any]: + try: + return self.operations[name] + except KeyError as exc: + raise ProfileError(f"Unsupported XDL operation: {name}") from exc + + +def _mapping(value: Any, name: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise ProfileError(f"{name} must be a mapping") + return value + + +def _handle_keys(action: dict[str, Any], io_type: str) -> set[str]: + handles = _mapping(action.get("handles", {}), "registry handles") + return { + str(item["handler_key"]) + for item in handles.get(io_type, []) + if isinstance(item, dict) and item.get("handler_key") + } + + +def load_station_profile(path: str | Path) -> StationProfile: + profile_path = Path(path).resolve() + raw = _mapping(yaml.safe_load(profile_path.read_text(encoding="utf-8")), "profile") + station = _mapping(raw.get("station"), "station") + hardware = _mapping(raw.get("hardware"), "hardware") + graph_path = (profile_path.parent / str(station["graph"])).resolve() + registry_path = (profile_path.parent / str(station["registry"])).resolve() + graph = json.loads(graph_path.read_text(encoding="utf-8")) + registry = _mapping( + yaml.safe_load(registry_path.read_text(encoding="utf-8")), "registry" + ) + graph_nodes = frozenset(str(node["id"]) for node in graph.get("nodes", [])) + workstation_id = str(station["workstation_id"]) + if workstation_id not in graph_nodes: + raise ProfileError(f"Workstation node does not exist: {workstation_id}") + mode = str(station.get("mode", "real")) + if mode not in {"real", "virtual"}: + raise ProfileError(f"Station mode must be real or virtual: {mode}") + + operations = standard_operations() + for name, override in _mapping(raw.get("operation_overrides", {}), "operation_overrides").items(): + if name not in operations: + raise ProfileError(f"Cannot override unknown operation: {name}") + operations[name].update(_mapping(override, f"operation_overrides.{name}")) + + resource_name = str(station["resource_name"]) + resource = _mapping(registry.get(resource_name), f"registry.{resource_name}") + actions = _mapping( + _mapping(resource.get("class"), f"registry.{resource_name}.class").get( + "action_value_mappings" + ), + "action_value_mappings", + ) + for name, operation in operations.items(): + action = actions.get(operation["template"]) + if not isinstance(action, dict): + continue + for handle in operation.get("inputs", {}).values(): + if handle not in _handle_keys(action, "input"): + raise ProfileError(f"{name}: missing input handle {handle}") + for handle in operation.get("outputs", {}).values(): + if handle not in _handle_keys(action, "output"): + raise ProfileError(f"{name}: missing output handle {handle}") + + return StationProfile( + path=profile_path, + graph_path=graph_path, + registry_path=registry_path, + workstation_id=workstation_id, + resource_name=resource_name, + mode=mode, + hardware_ids={str(k): str(v) for k, v in _mapping(hardware.get("ids", {}), "hardware.ids").items()}, + hardware_types={str(k): str(v) for k, v in _mapping(hardware.get("types", {}), "hardware.types").items()}, + operations=operations, + graph_nodes=graph_nodes, + )