diff --git a/unilabos/devices/workstation/bioyond_studio/dispensing_station.py b/unilabos/devices/workstation/bioyond_studio/dispensing_station.py index e80a1ef..a55a49c 100644 --- a/unilabos/devices/workstation/bioyond_studio/dispensing_station.py +++ b/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: - ratio = json.loads(ratio) - except Exception: - ratio = {} - root = str(Path(__file__).resolve().parents[3]) - if root not in sys.path: - sys.path.append(root) - try: - mod = importlib.import_module("tem.compute") - except Exception as e: - raise BioyondException(f"无法导入计算模块: {e}") + # 尝试解析简化格式: "MDA:0.5,PAPP:0.5,BTDA:0.95" + if ':' in ratio and ',' in ratio: + try: + ratio_dict = {} + pairs = ratio.split(',') + for pair in pairs: + pair = pair.strip() + if ':' in pair: + key, value = pair.split(':', 1) + key = key.strip() + 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, diff --git a/unilabos/registry/devices/bioyond_dispensing_station.yaml b/unilabos/registry/devices/bioyond_dispensing_station.yaml index db12e47..5ebad38 100644 --- a/unilabos/registry/devices/bioyond_dispensing_station.yaml +++ b/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: