mirror of
https://github.com/dptech-corp/Uni-Lab-OS.git
synced 2026-02-05 05:45:10 +00:00
acf7b6d3f7
* 修复了部分的protocol因为XDL更新导致的问题 但是pumptransfer，add，dissolve，separate还没修，后续还需要写virtual固体加料器 * 补充了四个action * 添加了固体加样器，丰富了json，修改了add protocol * bump version to 0.9.9 * fix bugs from new actions * protocol完整修复版本& bump version to 0.9.10 * 修补了一些单位处理，bump version to 0.9.11 * 优化了全protocol的运行时间，除了pumptransfer相关的还没 * 补充了剩下的几个protocol --------- Co-authored-by: Junhan Chang <changjh@dp.tech> Co-authored-by: Xuwznln <18435084+Xuwznln@users.noreply.github.com>
671 lines
28 KiB
Python
671 lines
28 KiB
Python
import networkx as nx
|
||
import re
|
||
import logging
|
||
import sys
|
||
from typing import List, Dict, Any, Union
|
||
from .pump_protocol import generate_pump_protocol_with_rinsing
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 确保输出编码为UTF-8
|
||
if hasattr(sys.stdout, 'reconfigure'):
|
||
try:
|
||
sys.stdout.reconfigure(encoding='utf-8')
|
||
sys.stderr.reconfigure(encoding='utf-8')
|
||
except:
|
||
pass
|
||
|
||
def debug_print(message):
|
||
"""调试输出函数 - 支持中文"""
|
||
try:
|
||
# 确保消息是字符串格式
|
||
safe_message = str(message)
|
||
print(f"[分离协议] {safe_message}", flush=True)
|
||
logger.info(f"[分离协议] {safe_message}")
|
||
except UnicodeEncodeError:
|
||
# 如果编码失败，尝试替换不支持的字符
|
||
safe_message = str(message).encode('utf-8', errors='replace').decode('utf-8')
|
||
print(f"[分离协议] {safe_message}", flush=True)
|
||
logger.info(f"[分离协议] {safe_message}")
|
||
except Exception as e:
|
||
# 最后的安全措施
|
||
fallback_message = f"日志输出错误: {repr(message)}"
|
||
print(f"[分离协议] {fallback_message}", flush=True)
|
||
logger.info(f"[分离协议] {fallback_message}")
|
||
|
||
def create_action_log(message: str, emoji: str = "📝") -> Dict[str, Any]:
|
||
"""创建一个动作日志 - 支持中文和emoji"""
|
||
try:
|
||
full_message = f"{emoji} {message}"
|
||
debug_print(full_message)
|
||
logger.info(full_message)
|
||
|
||
return {
|
||
"action_name": "wait",
|
||
"action_kwargs": {
|
||
"time": 0.1,
|
||
"log_message": full_message,
|
||
"progress_message": full_message
|
||
}
|
||
}
|
||
except Exception as e:
|
||
# 如果emoji有问题，使用纯文本
|
||
safe_message = f"[日志] {message}"
|
||
debug_print(safe_message)
|
||
logger.info(safe_message)
|
||
|
||
return {
|
||
"action_name": "wait",
|
||
"action_kwargs": {
|
||
"time": 0.1,
|
||
"log_message": safe_message,
|
||
"progress_message": safe_message
|
||
}
|
||
}
|
||
|
||
def parse_volume_input(volume_input: Union[str, float]) -> float:
|
||
"""
|
||
解析体积输入，支持带单位的字符串
|
||
|
||
Args:
|
||
volume_input: 体积输入（如 "200 mL", "?", 50.0）
|
||
|
||
Returns:
|
||
float: 体积（毫升）
|
||
"""
|
||
if isinstance(volume_input, (int, float)):
|
||
debug_print(f"📏 体积输入为数值: {volume_input}")
|
||
return float(volume_input)
|
||
|
||
if not volume_input or not str(volume_input).strip():
|
||
debug_print(f"⚠️ 体积输入为空，返回 0.0mL")
|
||
return 0.0
|
||
|
||
volume_str = str(volume_input).lower().strip()
|
||
debug_print(f"🔍 解析体积输入: '{volume_str}'")
|
||
|
||
# 处理未知体积
|
||
if volume_str in ['?', 'unknown', 'tbd', 'to be determined', '未知', '待定']:
|
||
default_volume = 100.0 # 默认100mL
|
||
debug_print(f"❓ 检测到未知体积，使用默认值: {default_volume}mL")
|
||
return default_volume
|
||
|
||
# 移除空格并提取数字和单位
|
||
volume_clean = re.sub(r'\s+', '', volume_str)
|
||
|
||
# 匹配数字和单位的正则表达式
|
||
match = re.match(r'([0-9]*\.?[0-9]+)\s*(ml|l|μl|ul|microliter|milliliter|liter|毫升|升|微升)?', volume_clean)
|
||
|
||
if not match:
|
||
debug_print(f"⚠️ 无法解析体积: '{volume_str}'，使用默认值 100mL")
|
||
return 100.0
|
||
|
||
value = float(match.group(1))
|
||
unit = match.group(2) or 'ml' # 默认单位为毫升
|
||
|
||
# 转换为毫升
|
||
if unit in ['l', 'liter', '升']:
|
||
volume = value * 1000.0 # L -> mL
|
||
debug_print(f"🔄 体积转换: {value}L -> {volume}mL")
|
||
elif unit in ['μl', 'ul', 'microliter', '微升']:
|
||
volume = value / 1000.0 # μL -> mL
|
||
debug_print(f"🔄 体积转换: {value}μL -> {volume}mL")
|
||
else: # ml, milliliter, 毫升 或默认
|
||
volume = value # 已经是mL
|
||
debug_print(f"✅ 体积已为毫升单位: {volume}mL")
|
||
|
||
return volume
|
||
|
||
def find_solvent_vessel(G: nx.DiGraph, solvent: str) -> str:
|
||
"""查找溶剂容器，支持多种匹配模式"""
|
||
if not solvent or not solvent.strip():
|
||
debug_print("⏭️ 未指定溶剂，跳过溶剂容器查找")
|
||
return ""
|
||
|
||
debug_print(f"🔍 正在查找溶剂 '{solvent}' 的容器...")
|
||
|
||
# 🔧 方法1：直接搜索 data.reagent_name 和 config.reagent
|
||
debug_print(f"📋 方法1: 搜索试剂字段...")
|
||
for node in G.nodes():
|
||
node_data = G.nodes[node].get('data', {})
|
||
node_type = G.nodes[node].get('type', '')
|
||
config_data = G.nodes[node].get('config', {})
|
||
|
||
# 只搜索容器类型的节点
|
||
if node_type == 'container':
|
||
reagent_name = node_data.get('reagent_name', '').lower()
|
||
config_reagent = config_data.get('reagent', '').lower()
|
||
|
||
# 精确匹配
|
||
if reagent_name == solvent.lower() or config_reagent == solvent.lower():
|
||
debug_print(f"✅ 通过试剂字段精确匹配找到容器: {node}")
|
||
return node
|
||
|
||
# 模糊匹配
|
||
if (solvent.lower() in reagent_name and reagent_name) or \
|
||
(solvent.lower() in config_reagent and config_reagent):
|
||
debug_print(f"✅ 通过试剂字段模糊匹配找到容器: {node}")
|
||
return node
|
||
|
||
# 🔧 方法2：常见的容器命名规则
|
||
debug_print(f"📋 方法2: 使用命名规则...")
|
||
solvent_clean = solvent.lower().replace(' ', '_').replace('-', '_')
|
||
possible_names = [
|
||
f"flask_{solvent_clean}",
|
||
f"bottle_{solvent_clean}",
|
||
f"vessel_{solvent_clean}",
|
||
f"{solvent_clean}_flask",
|
||
f"{solvent_clean}_bottle",
|
||
f"solvent_{solvent_clean}",
|
||
f"reagent_{solvent_clean}",
|
||
f"reagent_bottle_{solvent_clean}",
|
||
f"reagent_bottle_1", # 通用试剂瓶
|
||
f"reagent_bottle_2",
|
||
f"reagent_bottle_3"
|
||
]
|
||
|
||
debug_print(f"🎯 尝试的容器名称: {possible_names[:5]}... (共 {len(possible_names)} 个)")
|
||
|
||
for name in possible_names:
|
||
if name in G.nodes():
|
||
node_type = G.nodes[name].get('type', '')
|
||
if node_type == 'container':
|
||
debug_print(f"✅ 通过命名规则找到容器: {name}")
|
||
return name
|
||
|
||
# 🔧 方法3：使用第一个试剂瓶作为备选
|
||
debug_print(f"📋 方法3: 查找备用试剂瓶...")
|
||
for node_id in G.nodes():
|
||
node_data = G.nodes[node_id]
|
||
if (node_data.get('type') == 'container' and
|
||
('reagent' in node_id.lower() or 'bottle' in node_id.lower())):
|
||
debug_print(f"⚠️ 未找到专用容器，使用备用容器: {node_id}")
|
||
return node_id
|
||
|
||
debug_print(f"❌ 无法找到溶剂 '{solvent}' 的容器")
|
||
return ""
|
||
|
||
def find_separator_device(G: nx.DiGraph, vessel: str) -> str:
|
||
"""查找分离器设备，支持多种查找方式"""
|
||
debug_print(f"🔍 正在查找容器 '{vessel}' 的分离器设备...")
|
||
|
||
# 方法1：查找连接到容器的分离器设备
|
||
debug_print(f"📋 方法1: 检查连接的分离器...")
|
||
separator_nodes = []
|
||
for node in G.nodes():
|
||
node_class = G.nodes[node].get('class', '').lower()
|
||
if 'separator' in node_class:
|
||
separator_nodes.append(node)
|
||
debug_print(f"📋 发现分离器设备: {node}")
|
||
|
||
# 检查是否连接到目标容器
|
||
if G.has_edge(node, vessel) or G.has_edge(vessel, node):
|
||
debug_print(f"✅ 找到连接的分离器: {node}")
|
||
return node
|
||
|
||
debug_print(f"📊 找到的分离器总数: {len(separator_nodes)}")
|
||
|
||
# 方法2：根据命名规则查找
|
||
debug_print(f"📋 方法2: 使用命名规则...")
|
||
possible_names = [
|
||
f"{vessel}_controller",
|
||
f"{vessel}_separator",
|
||
vessel, # 容器本身可能就是分离器
|
||
"separator_1",
|
||
"virtual_separator",
|
||
"liquid_handler_1", # 液体处理器也可能用于分离
|
||
"controller_1"
|
||
]
|
||
|
||
debug_print(f"🎯 尝试的分离器名称: {possible_names}")
|
||
|
||
for name in possible_names:
|
||
if name in G.nodes():
|
||
node_class = G.nodes[name].get('class', '').lower()
|
||
if 'separator' in node_class or 'controller' in node_class:
|
||
debug_print(f"✅ 通过命名规则找到分离器: {name}")
|
||
return name
|
||
|
||
# 方法3：查找第一个分离器设备
|
||
debug_print(f"📋 方法3: 使用第一个可用分离器...")
|
||
if separator_nodes:
|
||
debug_print(f"⚠️ 使用第一个分离器设备: {separator_nodes[0]}")
|
||
return separator_nodes[0]
|
||
|
||
debug_print(f"❌ 未找到分离器设备")
|
||
return ""
|
||
|
||
def find_connected_stirrer(G: nx.DiGraph, vessel: str) -> str:
|
||
"""查找连接到指定容器的搅拌器"""
|
||
debug_print(f"🔍 正在查找与容器 {vessel} 连接的搅拌器...")
|
||
|
||
stirrer_nodes = []
|
||
for node in G.nodes():
|
||
node_data = G.nodes[node]
|
||
node_class = node_data.get('class', '') or ''
|
||
|
||
if 'stirrer' in node_class.lower():
|
||
stirrer_nodes.append(node)
|
||
debug_print(f"📋 发现搅拌器: {node}")
|
||
|
||
debug_print(f"📊 找到的搅拌器总数: {len(stirrer_nodes)}")
|
||
|
||
# 检查哪个搅拌器与目标容器相连
|
||
for stirrer in stirrer_nodes:
|
||
if G.has_edge(stirrer, vessel) or G.has_edge(vessel, stirrer):
|
||
debug_print(f"✅ 找到连接的搅拌器: {stirrer}")
|
||
return stirrer
|
||
|
||
# 如果没有连接的搅拌器，返回第一个可用的
|
||
if stirrer_nodes:
|
||
debug_print(f"⚠️ 未找到直接连接的搅拌器，使用第一个可用的: {stirrer_nodes[0]}")
|
||
return stirrer_nodes[0]
|
||
|
||
debug_print("❌ 未找到搅拌器")
|
||
return ""
|
||
|
||
def generate_separate_protocol(
|
||
G: nx.DiGraph,
|
||
# 🔧 基础参数，支持XDL的vessel参数
|
||
vessel: str = "", # XDL: 分离容器
|
||
purpose: str = "separate", # 分离目的
|
||
product_phase: str = "top", # 产物相
|
||
# 🔧 可选的详细参数
|
||
from_vessel: str = "", # 源容器（通常在separate前已经transfer了）
|
||
separation_vessel: str = "", # 分离容器（与vessel同义）
|
||
to_vessel: str = "", # 目标容器（可选）
|
||
waste_phase_to_vessel: str = "", # 废相目标容器
|
||
product_vessel: str = "", # XDL: 产物容器（与to_vessel同义）
|
||
waste_vessel: str = "", # XDL: 废液容器（与waste_phase_to_vessel同义）
|
||
# 🔧 溶剂相关参数
|
||
solvent: str = "", # 溶剂名称
|
||
solvent_volume: Union[str, float] = 0.0, # 溶剂体积
|
||
volume: Union[str, float] = 0.0, # XDL: 体积（与solvent_volume同义）
|
||
# 🔧 操作参数
|
||
through: str = "", # 通过材料
|
||
repeats: int = 1, # 重复次数
|
||
stir_time: float = 30.0, # 搅拌时间（秒）
|
||
stir_speed: float = 300.0, # 搅拌速度
|
||
settling_time: float = 300.0, # 沉降时间（秒）
|
||
**kwargs
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
生成分离操作的协议序列 - 增强中文版
|
||
|
||
支持XDL参数格式：
|
||
- vessel: 分离容器（必需）
|
||
- purpose: "wash", "extract", "separate"
|
||
- product_phase: "top", "bottom"
|
||
- product_vessel: 产物收集容器
|
||
- waste_vessel: 废液收集容器
|
||
- solvent: 溶剂名称
|
||
- volume: "200 mL", "?" 或数值
|
||
- repeats: 重复次数
|
||
|
||
分离流程：
|
||
1. （可选）添加溶剂到分离容器
|
||
2. 搅拌混合
|
||
3. 静置分层
|
||
4. 收集指定相到目标容器
|
||
5. 重复指定次数
|
||
"""
|
||
|
||
debug_print("=" * 60)
|
||
debug_print("🧪 开始生成分离协议 - 增强中文版")
|
||
debug_print(f"📋 原始参数:")
|
||
debug_print(f" 🥼 容器: '{vessel}'")
|
||
debug_print(f" 🎯 分离目的: '{purpose}'")
|
||
debug_print(f" 📊 产物相: '{product_phase}'")
|
||
debug_print(f" 💧 溶剂: '{solvent}'")
|
||
debug_print(f" 📏 体积: {volume} (类型: {type(volume)})")
|
||
debug_print(f" 🔄 重复次数: {repeats}")
|
||
debug_print(f" 🎯 产物容器: '{product_vessel}'")
|
||
debug_print(f" 🗑️ 废液容器: '{waste_vessel}'")
|
||
debug_print(f" 📦 其他参数: {kwargs}")
|
||
debug_print("=" * 60)
|
||
|
||
action_sequence = []
|
||
|
||
# === 参数验证和标准化 ===
|
||
debug_print("🔍 步骤1: 参数验证和标准化...")
|
||
action_sequence.append(create_action_log(f"开始分离操作 - 容器: {vessel}", "🎬"))
|
||
action_sequence.append(create_action_log(f"分离目的: {purpose}", "🧪"))
|
||
action_sequence.append(create_action_log(f"产物相: {product_phase}", "📊"))
|
||
|
||
# 统一容器参数
|
||
final_vessel = vessel or separation_vessel
|
||
if not final_vessel:
|
||
debug_print("❌ 必须指定分离容器")
|
||
raise ValueError("必须指定分离容器 (vessel 或 separation_vessel)")
|
||
|
||
final_to_vessel = to_vessel or product_vessel
|
||
final_waste_vessel = waste_phase_to_vessel or waste_vessel
|
||
|
||
# 统一体积参数
|
||
final_volume = parse_volume_input(volume or solvent_volume)
|
||
|
||
# 🔧 修复：确保repeats至少为1
|
||
if repeats <= 0:
|
||
repeats = 1
|
||
debug_print(f"⚠️ 重复次数参数 <= 0，自动设置为 1")
|
||
|
||
debug_print(f"🔧 标准化后的参数:")
|
||
debug_print(f" 🥼 分离容器: '{final_vessel}'")
|
||
debug_print(f" 🎯 产物容器: '{final_to_vessel}'")
|
||
debug_print(f" 🗑️ 废液容器: '{final_waste_vessel}'")
|
||
debug_print(f" 📏 溶剂体积: {final_volume}mL")
|
||
debug_print(f" 🔄 重复次数: {repeats}")
|
||
|
||
action_sequence.append(create_action_log(f"分离容器: {final_vessel}", "🧪"))
|
||
action_sequence.append(create_action_log(f"溶剂体积: {final_volume}mL", "📏"))
|
||
action_sequence.append(create_action_log(f"重复次数: {repeats}", "🔄"))
|
||
|
||
# 验证必需参数
|
||
if not purpose:
|
||
purpose = "separate"
|
||
if not product_phase:
|
||
product_phase = "top"
|
||
if purpose not in ["wash", "extract", "separate"]:
|
||
debug_print(f"⚠️ 未知的分离目的 '{purpose}'，使用默认值 'separate'")
|
||
purpose = "separate"
|
||
action_sequence.append(create_action_log(f"未知目的，使用: {purpose}", "⚠️"))
|
||
if product_phase not in ["top", "bottom"]:
|
||
debug_print(f"⚠️ 未知的产物相 '{product_phase}'，使用默认值 'top'")
|
||
product_phase = "top"
|
||
action_sequence.append(create_action_log(f"未知相别，使用: {product_phase}", "⚠️"))
|
||
|
||
debug_print("✅ 参数验证通过")
|
||
action_sequence.append(create_action_log("参数验证通过", "✅"))
|
||
|
||
# === 查找设备 ===
|
||
debug_print("🔍 步骤2: 查找设备...")
|
||
action_sequence.append(create_action_log("正在查找相关设备...", "🔍"))
|
||
|
||
# 查找分离器设备
|
||
separator_device = find_separator_device(G, final_vessel)
|
||
if separator_device:
|
||
action_sequence.append(create_action_log(f"找到分离器设备: {separator_device}", "🧪"))
|
||
else:
|
||
debug_print("⚠️ 未找到分离器设备，可能无法执行分离")
|
||
action_sequence.append(create_action_log("未找到分离器设备", "⚠️"))
|
||
|
||
# 查找搅拌器
|
||
stirrer_device = find_connected_stirrer(G, final_vessel)
|
||
if stirrer_device:
|
||
action_sequence.append(create_action_log(f"找到搅拌器: {stirrer_device}", "🌪️"))
|
||
else:
|
||
action_sequence.append(create_action_log("未找到搅拌器", "⚠️"))
|
||
|
||
# 查找溶剂容器（如果需要）
|
||
solvent_vessel = ""
|
||
if solvent and solvent.strip():
|
||
solvent_vessel = find_solvent_vessel(G, solvent)
|
||
if solvent_vessel:
|
||
action_sequence.append(create_action_log(f"找到溶剂容器: {solvent_vessel}", "💧"))
|
||
else:
|
||
action_sequence.append(create_action_log(f"未找到溶剂容器: {solvent}", "⚠️"))
|
||
|
||
debug_print(f"📊 设备配置:")
|
||
debug_print(f" 🧪 分离器设备: '{separator_device}'")
|
||
debug_print(f" 🌪️ 搅拌器设备: '{stirrer_device}'")
|
||
debug_print(f" 💧 溶剂容器: '{solvent_vessel}'")
|
||
|
||
# === 执行分离流程 ===
|
||
debug_print("🔍 步骤3: 执行分离流程...")
|
||
action_sequence.append(create_action_log("开始分离工作流程", "🎯"))
|
||
|
||
try:
|
||
for repeat_idx in range(repeats):
|
||
cycle_num = repeat_idx + 1
|
||
debug_print(f"🔄 第{cycle_num}轮: 开始分离循环 {cycle_num}/{repeats}")
|
||
action_sequence.append(create_action_log(f"分离循环 {cycle_num}/{repeats} 开始", "🔄"))
|
||
|
||
# 步骤3.1: 添加溶剂（如果需要）
|
||
if solvent_vessel and final_volume > 0:
|
||
debug_print(f"🔄 第{cycle_num}轮 步骤1: 添加溶剂 {solvent} ({final_volume}mL)")
|
||
action_sequence.append(create_action_log(f"向分离容器添加 {final_volume}mL {solvent}", "💧"))
|
||
|
||
try:
|
||
# 使用pump protocol添加溶剂
|
||
pump_actions = generate_pump_protocol_with_rinsing(
|
||
G=G,
|
||
from_vessel=solvent_vessel,
|
||
to_vessel=final_vessel,
|
||
volume=final_volume,
|
||
amount="",
|
||
time=0.0,
|
||
viscous=False,
|
||
rinsing_solvent="",
|
||
rinsing_volume=0.0,
|
||
rinsing_repeats=0,
|
||
solid=False,
|
||
flowrate=2.5,
|
||
transfer_flowrate=0.5,
|
||
rate_spec="",
|
||
event="",
|
||
through="",
|
||
**kwargs
|
||
)
|
||
action_sequence.extend(pump_actions)
|
||
debug_print(f"✅ 溶剂添加完成，添加了 {len(pump_actions)} 个动作")
|
||
action_sequence.append(create_action_log(f"溶剂转移完成 ({len(pump_actions)} 个操作)", "✅"))
|
||
|
||
except Exception as e:
|
||
debug_print(f"❌ 溶剂添加失败: {str(e)}")
|
||
action_sequence.append(create_action_log(f"溶剂添加失败: {str(e)}", "❌"))
|
||
else:
|
||
debug_print(f"🔄 第{cycle_num}轮 步骤1: 无需添加溶剂")
|
||
action_sequence.append(create_action_log("无需添加溶剂", "⏭️"))
|
||
|
||
# 步骤3.2: 启动搅拌（如果有搅拌器）
|
||
if stirrer_device and stir_time > 0:
|
||
debug_print(f"🔄 第{cycle_num}轮 步骤2: 开始搅拌 ({stir_speed}rpm，持续 {stir_time}s)")
|
||
action_sequence.append(create_action_log(f"开始搅拌: {stir_speed}rpm，持续 {stir_time}s", "🌪️"))
|
||
|
||
action_sequence.append({
|
||
"device_id": stirrer_device,
|
||
"action_name": "start_stir",
|
||
"action_kwargs": {
|
||
"vessel": final_vessel,
|
||
"stir_speed": stir_speed,
|
||
"purpose": f"分离混合 - {purpose}"
|
||
}
|
||
})
|
||
|
||
# 搅拌等待
|
||
stir_minutes = stir_time / 60
|
||
action_sequence.append(create_action_log(f"搅拌中，持续 {stir_minutes:.1f} 分钟", "⏱️"))
|
||
action_sequence.append({
|
||
"action_name": "wait",
|
||
"action_kwargs": {"time": stir_time}
|
||
})
|
||
|
||
# 停止搅拌
|
||
action_sequence.append(create_action_log("停止搅拌器", "🛑"))
|
||
action_sequence.append({
|
||
"device_id": stirrer_device,
|
||
"action_name": "stop_stir",
|
||
"action_kwargs": {"vessel": final_vessel}
|
||
})
|
||
|
||
else:
|
||
debug_print(f"🔄 第{cycle_num}轮 步骤2: 无需搅拌")
|
||
action_sequence.append(create_action_log("无需搅拌", "⏭️"))
|
||
|
||
# 步骤3.3: 静置分层
|
||
if settling_time > 0:
|
||
debug_print(f"🔄 第{cycle_num}轮 步骤3: 静置分层 ({settling_time}s)")
|
||
settling_minutes = settling_time / 60
|
||
action_sequence.append(create_action_log(f"静置分层 ({settling_minutes:.1f} 分钟)", "⚖️"))
|
||
action_sequence.append({
|
||
"action_name": "wait",
|
||
"action_kwargs": {"time": settling_time}
|
||
})
|
||
else:
|
||
debug_print(f"🔄 第{cycle_num}轮 步骤3: 未指定静置时间")
|
||
action_sequence.append(create_action_log("未指定静置时间", "⏭️"))
|
||
|
||
# 步骤3.4: 执行分离操作
|
||
if separator_device:
|
||
debug_print(f"🔄 第{cycle_num}轮 步骤4: 执行分离操作")
|
||
action_sequence.append(create_action_log(f"执行分离: 收集{product_phase}相", "🧪"))
|
||
|
||
# 调用分离器设备的separate方法
|
||
separate_action = {
|
||
"device_id": separator_device,
|
||
"action_name": "separate",
|
||
"action_kwargs": {
|
||
"purpose": purpose,
|
||
"product_phase": product_phase,
|
||
"from_vessel": from_vessel or final_vessel,
|
||
"separation_vessel": final_vessel,
|
||
"to_vessel": final_to_vessel or final_vessel,
|
||
"waste_phase_to_vessel": final_waste_vessel or final_vessel,
|
||
"solvent": solvent,
|
||
"solvent_volume": final_volume,
|
||
"through": through,
|
||
"repeats": 1, # 每次调用只做一次分离
|
||
"stir_time": 0, # 已经在上面完成
|
||
"stir_speed": stir_speed,
|
||
"settling_time": 0 # 已经在上面完成
|
||
}
|
||
}
|
||
action_sequence.append(separate_action)
|
||
debug_print(f"✅ 分离操作已添加")
|
||
action_sequence.append(create_action_log("分离操作完成", "✅"))
|
||
|
||
# 收集结果
|
||
if final_to_vessel:
|
||
action_sequence.append(create_action_log(f"产物 ({product_phase}相) 收集到: {final_to_vessel}", "📦"))
|
||
if final_waste_vessel:
|
||
action_sequence.append(create_action_log(f"废相收集到: {final_waste_vessel}", "🗑️"))
|
||
|
||
else:
|
||
debug_print(f"🔄 第{cycle_num}轮 步骤4: 无分离器设备，跳过分离")
|
||
action_sequence.append(create_action_log("无分离器设备可用", "❌"))
|
||
# 添加等待时间模拟分离
|
||
action_sequence.append({
|
||
"action_name": "wait",
|
||
"action_kwargs": {"time": 10.0}
|
||
})
|
||
|
||
# 循环间等待（除了最后一次）
|
||
if repeat_idx < repeats - 1:
|
||
debug_print(f"🔄 第{cycle_num}轮: 等待下一次循环...")
|
||
action_sequence.append(create_action_log("等待下一次循环...", "⏳"))
|
||
action_sequence.append({
|
||
"action_name": "wait",
|
||
"action_kwargs": {"time": 5}
|
||
})
|
||
else:
|
||
action_sequence.append(create_action_log(f"分离循环 {cycle_num}/{repeats} 完成", "🌟"))
|
||
|
||
except Exception as e:
|
||
debug_print(f"❌ 分离工作流程执行失败: {str(e)}")
|
||
action_sequence.append(create_action_log(f"分离工作流程失败: {str(e)}", "❌"))
|
||
# 添加错误日志
|
||
action_sequence.append({
|
||
"device_id": "system",
|
||
"action_name": "log_message",
|
||
"action_kwargs": {
|
||
"message": f"分离操作失败: {str(e)}"
|
||
}
|
||
})
|
||
|
||
# === 最终结果 ===
|
||
total_time = (stir_time + settling_time + 15) * repeats # 估算总时间
|
||
|
||
debug_print("=" * 60)
|
||
debug_print(f"🎉 分离协议生成完成")
|
||
debug_print(f"📊 协议统计:")
|
||
debug_print(f" 📋 总动作数: {len(action_sequence)}")
|
||
debug_print(f" ⏱️ 预计总时间: {total_time:.0f}s ({total_time/60:.1f} 分钟)")
|
||
debug_print(f" 🥼 分离容器: {final_vessel}")
|
||
debug_print(f" 🎯 分离目的: {purpose}")
|
||
debug_print(f" 📊 产物相: {product_phase}")
|
||
debug_print(f" 🔄 重复次数: {repeats}")
|
||
if solvent:
|
||
debug_print(f" 💧 溶剂: {solvent} ({final_volume}mL)")
|
||
if final_to_vessel:
|
||
debug_print(f" 🎯 产物容器: {final_to_vessel}")
|
||
if final_waste_vessel:
|
||
debug_print(f" 🗑️ 废液容器: {final_waste_vessel}")
|
||
debug_print("=" * 60)
|
||
|
||
# 添加完成日志
|
||
summary_msg = f"分离协议完成: {final_vessel} ({purpose}，{repeats} 次循环)"
|
||
if solvent:
|
||
summary_msg += f"，使用 {final_volume}mL {solvent}"
|
||
action_sequence.append(create_action_log(summary_msg, "🎉"))
|
||
|
||
return action_sequence
|
||
|
||
# === 便捷函数 ===
|
||
|
||
def separate_phases_only(G: nx.DiGraph, vessel: str, product_phase: str = "top",
|
||
product_vessel: str = "", waste_vessel: str = "") -> List[Dict[str, Any]]:
|
||
"""仅进行相分离（不添加溶剂）"""
|
||
debug_print(f"⚡ 快速相分离: {vessel} ({product_phase}相)")
|
||
return generate_separate_protocol(
|
||
G, vessel=vessel,
|
||
purpose="separate",
|
||
product_phase=product_phase,
|
||
product_vessel=product_vessel,
|
||
waste_vessel=waste_vessel
|
||
)
|
||
|
||
def wash_with_solvent(G: nx.DiGraph, vessel: str, solvent: str, volume: Union[str, float],
|
||
product_phase: str = "top", repeats: int = 1) -> List[Dict[str, Any]]:
|
||
"""用溶剂洗涤"""
|
||
debug_print(f"🧽 用{solvent}洗涤: {vessel} ({repeats} 次)")
|
||
return generate_separate_protocol(
|
||
G, vessel=vessel,
|
||
purpose="wash",
|
||
product_phase=product_phase,
|
||
solvent=solvent,
|
||
volume=volume,
|
||
repeats=repeats
|
||
)
|
||
|
||
def extract_with_solvent(G: nx.DiGraph, vessel: str, solvent: str, volume: Union[str, float],
|
||
product_phase: str = "bottom", repeats: int = 3) -> List[Dict[str, Any]]:
|
||
"""用溶剂萃取"""
|
||
debug_print(f"🧪 用{solvent}萃取: {vessel} ({repeats} 次)")
|
||
return generate_separate_protocol(
|
||
G, vessel=vessel,
|
||
purpose="extract",
|
||
product_phase=product_phase,
|
||
solvent=solvent,
|
||
volume=volume,
|
||
repeats=repeats
|
||
)
|
||
|
||
def separate_aqueous_organic(G: nx.DiGraph, vessel: str, organic_phase: str = "top",
|
||
product_vessel: str = "", waste_vessel: str = "") -> List[Dict[str, Any]]:
|
||
"""水-有机相分离"""
|
||
debug_print(f"💧 水-有机相分离: {vessel} (有机相: {organic_phase})")
|
||
return generate_separate_protocol(
|
||
G, vessel=vessel,
|
||
purpose="separate",
|
||
product_phase=organic_phase,
|
||
product_vessel=product_vessel,
|
||
waste_vessel=waste_vessel
|
||
)
|
||
|
||
# 测试函数
|
||
def test_separate_protocol():
|
||
"""测试分离协议的各种参数解析"""
|
||
debug_print("=== 分离协议增强中文版测试 ===")
|
||
|
||
# 测试体积解析
|
||
debug_print("🧪 测试体积解析...")
|
||
volumes = ["200 mL", "?", 100.0, "1 L", "500 μL", "未知", "50毫升"]
|
||
for vol in volumes:
|
||
result = parse_volume_input(vol)
|
||
debug_print(f"📊 体积解析结果: {vol} -> {result}mL")
|
||
|
||
debug_print("✅ 测试完成")
|
||
|
||
if __name__ == "__main__":
|
||
test_separate_protocol()
|