feat(bioyond): 添加计算实验设计功能，支持化合物配比和滴定比例参数

unilabos/devices/workstation/bioyond_studio/dispensing_station.py

@@ -3,6 +3,7 @@ import json
 import time
 from typing import Optional, Dict, Any, List
 import requests
+import pint
 from unilabos.devices.workstation.bioyond_studio.config import API_CONFIG
 
 from unilabos.devices.workstation.bioyond_studio.bioyond_rpc import BioyondException
@@ -34,6 +35,41 @@ class BioyondDispensingStation(BioyondWorkstation):
         # 用于跟踪任务完成状态的字典: {orderCode: {status, order_id, timestamp}}
         self.order_completion_status = {}
 
+        # 初始化 pint 单位注册表
+        self.ureg = pint.UnitRegistry()
+
+        # 化合物信息
+        self.compound_info = {
+            "MolWt": {
+                "MDA": 108.14 * self.ureg.g / self.ureg.mol,
+                "TDA": 122.16 * self.ureg.g / self.ureg.mol,
+                "PAPP": 521.62 * self.ureg.g / self.ureg.mol,
+                "BTDA": 322.23 * self.ureg.g / self.ureg.mol,
+                "BPDA": 294.22 * self.ureg.g / self.ureg.mol,
+                "6FAP": 366.26 * self.ureg.g / self.ureg.mol,
+                "PMDA": 218.12 * self.ureg.g / self.ureg.mol,
+                "MPDA": 108.14 * self.ureg.g / self.ureg.mol,
+                "SIDA": 248.51 * self.ureg.g / self.ureg.mol,
+                "ODA": 200.236 * self.ureg.g / self.ureg.mol,
+                "4,4'-ODA": 200.236 * self.ureg.g / self.ureg.mol,
+                "134": 292.34 * self.ureg.g / self.ureg.mol,
+            },
+            "FuncGroup": {
+                "MDA": "Amine",
+                "TDA": "Amine",
+                "PAPP": "Amine",
+                "BTDA": "Anhydride",
+                "BPDA": "Anhydride",
+                "6FAP": "Amine",
+                "MPDA": "Amine",
+                "SIDA": "Amine",
+                "PMDA": "Anhydride",
+                "ODA": "Amine",
+                "4,4'-ODA": "Amine",
+                "134": "Amine",
+            }
+        }
+
     def _post_project_api(self, endpoint: str, data: Any) -> Dict[str, Any]:
         """项目接口通用POST调用
 
@@ -103,26 +139,100 @@ class BioyondDispensingStation(BioyondWorkstation):
         m_tot: str = "70",
         titration_percent: str = "0.03",
     ) -> dict:
+        """计算实验设计参数
+
+        参数:
+            ratio: 化合物配比，支持多种格式:
+                   1. 简化格式(推荐): "MDA:0.5,PAPP:0.5,BTDA:0.95"
+                   2. JSON字符串: '{"MDA": 1, "BTDA": 0.95, "PAPP": 1}'
+                   3. Python字典: {"MDA": 1, "BTDA": 0.95, "PAPP": 1}
+            wt_percent: 固体重量百分比，默认 0.25
+            m_tot: 反应混合物总质量(g)，默认 70
+            titration_percent: 滴定溶液百分比，默认 0.03
+
+        返回:
+            包含实验设计参数的字典
+        """
         try:
+            # 1. 参数解析和验证
+            original_ratio = ratio
+
             if isinstance(ratio, str):
-                try:
+                # 尝试解析简化格式: "MDA:0.5,PAPP:0.5,BTDA:0.95"
-                    ratio = json.loads(ratio)
+                if ':' in ratio and ',' in ratio:
-                except Exception:
+                    try:
-                    ratio = {}
+                        ratio_dict = {}
-            root = str(Path(__file__).resolve().parents[3])
+                        pairs = ratio.split(',')
-            if root not in sys.path:
+                        for pair in pairs:
-                sys.path.append(root)
+                            pair = pair.strip()
-            try:
+                            if ':' in pair:
-                mod = importlib.import_module("tem.compute")
+                                key, value = pair.split(':', 1)
-            except Exception as e:
+                                key = key.strip()
-                raise BioyondException(f"无法导入计算模块: {e}")
+                                value = value.strip()
+                                try:
+                                    ratio_dict[key] = float(value)
+                                except ValueError:
+                                    raise BioyondException(f"无法将 '{value}' 转换为数字")
+                        if ratio_dict:
+                            ratio = ratio_dict
+                            self.hardware_interface._logger.info(
+                                f"从简化格式解析 ratio: '{original_ratio}' -> {ratio}"
+                            )
+                    except BioyondException:
+                        raise
+                    except Exception as e:
+                        self.hardware_interface._logger.warning(
+                            f"简化格式解析失败，尝试JSON格式: {e}"
+                        )
+
+                # 如果不是简化格式或解析失败，尝试JSON格式
+                if isinstance(ratio, str):
+                    try:
+                        ratio = json.loads(ratio)
+                        # 处理可能的多层 JSON 编码
+                        if isinstance(ratio, str):
+                            try:
+                                ratio = json.loads(ratio)
+                            except Exception:
+                                pass
+                    except Exception as e:
+                        raise BioyondException(
+                            f"ratio 参数解析失败: {e}。\n"
+                            f"支持的格式:\n"
+                            f"  1. 简化格式(推荐): 'MDA:0.5,PAPP:0.5,BTDA:0.95'\n"
+                            f"  2. JSON格式: '{{\"MDA\": 0.5, \"BTDA\": 0.95, \"PAPP\": 0.5}}'"
+                        )
+
+            if not isinstance(ratio, dict):
+                raise BioyondException(
+                    f"ratio 必须是字典类型或可解析的字符串，当前类型: {type(ratio)}。\n"
+                    f"支持的格式:\n"
+                    f"  1. 简化格式(推荐): 'MDA:0.5,PAPP:0.5,BTDA:0.95'\n"
+                    f"  2. JSON格式: '{{\"MDA\": 0.5, \"BTDA\": 0.95, \"PAPP\": 0.5}}'"
+                )
+
+            if not ratio:
+                raise BioyondException("ratio 参数不能为空")
+
+            # 记录解析后的参数用于调试
+            self.hardware_interface._logger.info(f"最终解析的 ratio 参数: {ratio} (类型: {type(ratio)})")
+
             try:
                 wp = float(wt_percent) if isinstance(wt_percent, str) else wt_percent
                 mt = float(m_tot) if isinstance(m_tot, str) else m_tot
                 tp = float(titration_percent) if isinstance(titration_percent, str) else titration_percent
             except Exception as e:
                 raise BioyondException(f"参数解析失败: {e}")
-            res = mod.generate_experiment_design(ratio=ratio, wt_percent=wp, m_tot=mt, titration_percent=tp)
+
+            # 2. 调用内部计算方法
+            res = self._generate_experiment_design(
+                ratio=ratio,
+                wt_percent=wp,
+                m_tot=mt,
+                titration_percent=tp
+            )
+
+            # 3. 构造返回结果
             out = {
                 "solutions": res.get("solutions", []),
                 "titration": res.get("titration", {}),
@@ -131,11 +241,248 @@ class BioyondDispensingStation(BioyondWorkstation):
                 "return_info": json.dumps(res, ensure_ascii=False)
             }
             return out
+
         except BioyondException:
             raise
         except Exception as e:
             raise BioyondException(str(e))
 
+    def _generate_experiment_design(
+        self,
+        ratio: dict,
+        wt_percent: float = 0.25,
+        m_tot: float = 70,
+        titration_percent: float = 0.03,
+    ) -> dict:
+        """内部方法：生成实验设计
+
+        根据FuncGroup自动区分二胺和二酐，每种二胺单独配溶液，严格按照ratio顺序投料。
+
+        参数:
+            ratio: 化合物配比字典，格式: {"compound_name": ratio_value}
+            wt_percent: 固体重量百分比
+            m_tot: 反应混合物总质量(g)
+            titration_percent: 滴定溶液百分比
+
+        返回:
+            包含实验设计详细参数的字典
+        """
+        # 溶剂密度
+        ρ_solvent = 1.03 * self.ureg.g / self.ureg.ml
+        # 二酐溶解度
+        solubility = 0.02 * self.ureg.g / self.ureg.ml
+        # 投入固体时最小溶剂体积
+        V_min = 30 * self.ureg.ml
+        m_tot = m_tot * self.ureg.g
+
+        # 保持ratio中的顺序
+        compound_names = list(ratio.keys())
+        compound_ratios = list(ratio.values())
+
+        # 验证所有化合物是否在 compound_info 中定义
+        undefined_compounds = [name for name in compound_names if name not in self.compound_info["MolWt"]]
+        if undefined_compounds:
+            available = list(self.compound_info["MolWt"].keys())
+            raise ValueError(
+                f"以下化合物未在 compound_info 中定义: {undefined_compounds}。"
+                f"可用的化合物: {available}"
+            )
+
+        # 获取各化合物的分子量和官能团类型
+        molecular_weights = [self.compound_info["MolWt"][name] for name in compound_names]
+        func_groups = [self.compound_info["FuncGroup"][name] for name in compound_names]
+
+        # 记录化合物信息用于调试
+        self.hardware_interface._logger.info(f"化合物名称: {compound_names}")
+        self.hardware_interface._logger.info(f"官能团类型: {func_groups}")
+
+        # 按原始顺序分离二胺和二酐
+        ordered_compounds = list(zip(compound_names, compound_ratios, molecular_weights, func_groups))
+        diamine_compounds = [(name, ratio_val, mw, i) for i, (name, ratio_val, mw, fg) in enumerate(ordered_compounds) if fg == "Amine"]
+        anhydride_compounds = [(name, ratio_val, mw, i) for i, (name, ratio_val, mw, fg) in enumerate(ordered_compounds) if fg == "Anhydride"]
+
+        if not diamine_compounds or not anhydride_compounds:
+            raise ValueError(
+                f"需要同时包含二胺(Amine)和二酐(Anhydride)化合物。"
+                f"当前二胺: {[c[0] for c in diamine_compounds]}, "
+                f"当前二酐: {[c[0] for c in anhydride_compounds]}"
+            )
+
+        # 计算加权平均分子量 (基于摩尔比)
+        total_molar_ratio = sum(compound_ratios)
+        weighted_molecular_weight = sum(ratio_val * mw for ratio_val, mw in zip(compound_ratios, molecular_weights))
+
+        # 取最后一个二酐用于滴定
+        titration_anhydride = anhydride_compounds[-1]
+        solid_anhydrides = anhydride_compounds[:-1] if len(anhydride_compounds) > 1 else []
+
+        # 二胺溶液配制参数 - 每种二胺单独配制
+        diamine_solutions = []
+        total_diamine_volume = 0 * self.ureg.ml
+
+        # 计算反应物的总摩尔量
+        n_reactant = m_tot * wt_percent / weighted_molecular_weight
+
+        for name, ratio_val, mw, order_index in diamine_compounds:
+            # 跳过 SIDA
+            if name == "SIDA":
+                continue
+
+            # 计算该二胺需要的摩尔数
+            n_diamine_needed = n_reactant * ratio_val
+
+            # 二胺溶液配制参数 (每种二胺固定配制参数)
+            m_diamine_solid = 5.0 * self.ureg.g  # 每种二胺固体质量
+            V_solvent_for_this = 20 * self.ureg.ml  # 每种二胺溶剂体积
+            m_solvent_for_this = ρ_solvent * V_solvent_for_this
+
+            # 计算该二胺溶液的浓度
+            c_diamine = (m_diamine_solid / mw) / V_solvent_for_this
+
+            # 计算需要移取的溶液体积
+            V_diamine_needed = n_diamine_needed / c_diamine
+
+            diamine_solutions.append({
+                "name": name,
+                "order": order_index,
+                "solid_mass": m_diamine_solid.magnitude,
+                "solvent_volume": V_solvent_for_this.magnitude,
+                "concentration": c_diamine.magnitude,
+                "volume_needed": V_diamine_needed.magnitude,
+                "molar_ratio": ratio_val
+            })
+
+            total_diamine_volume += V_diamine_needed
+
+        # 按原始顺序排序
+        diamine_solutions.sort(key=lambda x: x["order"])
+
+        # 计算滴定二酐的质量
+        titration_name, titration_ratio, titration_mw, _ = titration_anhydride
+        m_titration_anhydride = n_reactant * titration_ratio * titration_mw
+        m_titration_90 = m_titration_anhydride * (1 - titration_percent)
+        m_titration_10 = m_titration_anhydride * titration_percent
+
+        # 计算其他固体二酐的质量 (按顺序)
+        solid_anhydride_masses = []
+        for name, ratio_val, mw, order_index in solid_anhydrides:
+            mass = n_reactant * ratio_val * mw
+            solid_anhydride_masses.append({
+                "name": name,
+                "order": order_index,
+                "mass": mass.magnitude,
+                "molar_ratio": ratio_val
+            })
+
+        # 按原始顺序排序
+        solid_anhydride_masses.sort(key=lambda x: x["order"])
+
+        # 计算溶剂用量
+        total_diamine_solution_mass = sum(
+            sol["volume_needed"] * ρ_solvent for sol in diamine_solutions
+        ) * self.ureg.ml
+
+        # 预估滴定溶剂量、计算补加溶剂量
+        m_solvent_titration = m_titration_10 / solubility * ρ_solvent
+        m_solvent_add = m_tot * (1 - wt_percent) - total_diamine_solution_mass - m_solvent_titration
+
+        # 检查最小溶剂体积要求
+        total_liquid_volume = (total_diamine_solution_mass + m_solvent_add) / ρ_solvent
+        m_tot_min = V_min / total_liquid_volume * m_tot
+
+        # 如果需要，按比例放大
+        scale_factor = 1.0
+        if m_tot_min > m_tot:
+            scale_factor = (m_tot_min / m_tot).magnitude
+            m_titration_90 *= scale_factor
+            m_titration_10 *= scale_factor
+            m_solvent_add *= scale_factor
+            m_solvent_titration *= scale_factor
+
+            # 更新二胺溶液用量
+            for sol in diamine_solutions:
+                sol["volume_needed"] *= scale_factor
+
+            # 更新固体二酐用量
+            for anhydride in solid_anhydride_masses:
+                anhydride["mass"] *= scale_factor
+
+            m_tot = m_tot_min
+
+        # 生成投料顺序
+        feeding_order = []
+
+        # 1. 固体二酐 (按顺序)
+        for anhydride in solid_anhydride_masses:
+            feeding_order.append({
+                "step": len(feeding_order) + 1,
+                "type": "solid_anhydride",
+                "name": anhydride["name"],
+                "amount": anhydride["mass"],
+                "order": anhydride["order"]
+            })
+
+        # 2. 二胺溶液 (按顺序)
+        for sol in diamine_solutions:
+            feeding_order.append({
+                "step": len(feeding_order) + 1,
+                "type": "diamine_solution",
+                "name": sol["name"],
+                "amount": sol["volume_needed"],
+                "order": sol["order"]
+            })
+
+        # 3. 主要二酐粉末
+        feeding_order.append({
+            "step": len(feeding_order) + 1,
+            "type": "main_anhydride",
+            "name": titration_name,
+            "amount": m_titration_90.magnitude,
+            "order": titration_anhydride[3]
+        })
+
+        # 4. 补加溶剂
+        if m_solvent_add > 0:
+            feeding_order.append({
+                "step": len(feeding_order) + 1,
+                "type": "additional_solvent",
+                "name": "溶剂",
+                "amount": m_solvent_add.magnitude,
+                "order": 999
+            })
+
+        # 5. 滴定二酐溶液
+        feeding_order.append({
+            "step": len(feeding_order) + 1,
+            "type": "titration_anhydride",
+            "name": f"{titration_name} 滴定液",
+            "amount": m_titration_10.magnitude,
+            "titration_solvent": m_solvent_titration.magnitude,
+            "order": titration_anhydride[3]
+        })
+
+        # 返回实验设计结果
+        results = {
+            "total_mass": m_tot.magnitude,
+            "scale_factor": scale_factor,
+            "solutions": diamine_solutions,
+            "solids": solid_anhydride_masses,
+            "titration": {
+                "name": titration_name,
+                "main_portion": m_titration_90.magnitude,
+                "titration_portion": m_titration_10.magnitude,
+                "titration_solvent": m_solvent_titration.magnitude,
+            },
+            "solvents": {
+                "additional_solvent": m_solvent_add.magnitude,
+                "total_liquid_volume": total_liquid_volume.magnitude
+            },
+            "feeding_order": feeding_order,
+            "minimum_required_mass": m_tot_min.magnitude
+        }
+
+        return results
+
     # 90%10%小瓶投料任务创建方法
     def create_90_10_vial_feeding_task(self,
                                        order_name: str = None,

unilabos/registry/devices/bioyond_dispensing_station.yaml

@@ -200,6 +200,99 @@ bioyond_dispensing_station:
           title: BatchCreateDiamineSolutionTasks
           type: object
         type: UniLabJsonCommand
+      compute_experiment_design:
+        feedback: {}
+        goal:
+          m_tot: m_tot
+          ratio: ratio
+          titration_percent: titration_percent
+          wt_percent: wt_percent
+        goal_default:
+          m_tot: '70'
+          ratio: ''
+          titration_percent: '0.03'
+          wt_percent: '0.25'
+        handles:
+          output:
+          - data_key: solutions
+            data_source: executor
+            data_type: array
+            handler_key: solutions
+            io_type: sink
+            label: Solution Data From Python
+          - data_key: titration
+            data_source: executor
+            data_type: object
+            handler_key: titration
+            io_type: sink
+            label: Titration Data From Calculation Node
+          - data_key: solvents
+            data_source: executor
+            data_type: object
+            handler_key: solvents
+            io_type: sink
+            label: Solvents Data From Calculation Node
+          - data_key: feeding_order
+            data_source: executor
+            data_type: array
+            handler_key: feeding_order
+            io_type: sink
+            label: Feeding Order Data From Calculation Node
+        result:
+          feeding_order: feeding_order
+          return_info: return_info
+          solutions: solutions
+          solvents: solvents
+          titration: titration
+        schema:
+          description: 计算实验设计，输出solutions/titration/solvents/feeding_order用于后续节点。
+          properties:
+            feedback: {}
+            goal:
+              properties:
+                m_tot:
+                  default: '70'
+                  description: 总质量(g)
+                  type: string
+                ratio:
+                  description: 组分摩尔比的对象，保持输入顺序，如{"MDA":1,"BTDA":1}
+                  type: string
+                titration_percent:
+                  default: '0.03'
+                  description: 滴定比例(10%部分)
+                  type: string
+                wt_percent:
+                  default: '0.25'
+                  description: 目标固含质量分数
+                  type: string
+              required:
+              - ratio
+              type: object
+            result:
+              properties:
+                feeding_order:
+                  type: array
+                return_info:
+                  type: string
+                solutions:
+                  type: array
+                solvents:
+                  type: object
+                titration:
+                  type: object
+              required:
+              - solutions
+              - titration
+              - solvents
+              - feeding_order
+              - return_info
+              title: ComputeExperimentDesign_Result
+              type: object
+          required:
+          - goal
+          title: ComputeExperimentDesign
+          type: object
+        type: UniLabJsonCommand
       create_90_10_vial_feeding_task:
         feedback: {}
         goal: