From d75c7f123b0ca5576117ec62ad895767c0100600 Mon Sep 17 00:00:00 2001 From: Junhan Chang Date: Wed, 25 Mar 2026 13:11:14 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=8D=87=E7=BA=A7=E7=BC=96?= =?UTF-8?q?=E8=AF=91=E5=99=A8=E5=85=B1=E4=BA=AB=E5=B7=A5=E5=85=B7=E5=BA=93?= =?UTF-8?q?=EF=BC=88logger=5Futil,=20unit=5Fparser,=20vessel=5Fparser,=20r?= =?UTF-8?q?esource=5Fhelper=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - logger_util: 重写debug_print，支持自动检测调用模块并设置前缀 - unit_parser: 新增parse_temperature_input，统一温度字符串解析 - vessel_parser: 新增find_connected_heatchill，统一加热设备查找 - resource_helper: 新增update_vessel_volume/get_resource_liquid_volume等共享函数 Co-Authored-By: Claude Opus 4.6 --- unilabos/compile/utils/logger_util.py | 75 ++++++---- unilabos/compile/utils/resource_helper.py | 172 ++++++++++++++++++++++ unilabos/compile/utils/unit_parser.py | 36 +++++ unilabos/compile/utils/vessel_parser.py | 45 ++++-- 4 files changed, 290 insertions(+), 38 deletions(-) create mode 100644 unilabos/compile/utils/resource_helper.py diff --git a/unilabos/compile/utils/logger_util.py b/unilabos/compile/utils/logger_util.py index 635e11e2..b73c8674 100644 --- a/unilabos/compile/utils/logger_util.py +++ b/unilabos/compile/utils/logger_util.py @@ -1,36 +1,57 @@ -# 🆕 创建进度日志动作 +"""编译器共享日志工具""" + +import inspect import logging from typing import Dict, Any -logger = logging.getLogger(__name__) +# 模块名到前缀的映射 +_MODULE_PREFIXES = { + "add_protocol": "[ADD]", + "adjustph_protocol": "[ADJUSTPH]", + "clean_vessel_protocol": "[CLEAN_VESSEL]", + "dissolve_protocol": "[DISSOLVE]", + "dry_protocol": "[DRY]", + "evacuateandrefill_protocol": "[EVACUATE]", + "evaporate_protocol": "[EVAPORATE]", + "filter_protocol": "[FILTER]", + "heatchill_protocol": "[HEATCHILL]", + "hydrogenate_protocol": "[HYDROGENATE]", + "pump_protocol": "[PUMP]", + "recrystallize_protocol": "[RECRYSTALLIZE]", + "reset_handling_protocol": "[RESET]", + "run_column_protocol": "[RUN_COLUMN]", + "separate_protocol": "[SEPARATE]", + "stir_protocol": "[STIR]", + "wash_solid_protocol": "[WASH_SOLID]", + "vessel_parser": "[VESSEL_PARSER]", + "unit_parser": "[UNIT_PARSER]", + "resource_helper": "[RESOURCE_HELPER]", +} -def debug_print(message, prefix="[UNIT_PARSER]"): - """调试输出""" + +def debug_print(message, prefix=None): + """调试输出 — 自动根据调用模块设置前缀""" + if prefix is None: + frame = inspect.currentframe() + caller = frame.f_back if frame else None + module_name = "" + if caller: + module_name = caller.f_globals.get("__name__", "") + # 取最后一段作为模块短名 + module_name = module_name.rsplit(".", 1)[-1] + prefix = _MODULE_PREFIXES.get(module_name, f"[{module_name.upper()}]") + logger = logging.getLogger("unilabos.compile") logger.info(f"{prefix} {message}") def action_log(message: str, emoji: str = "📝", prefix="[HIGH-LEVEL OPERATION]") -> Dict[str, Any]: - """创建一个动作日志 - 支持中文和emoji""" - try: - full_message = f"{prefix} {emoji} {message}" - - return { - "action_name": "wait", - "action_kwargs": { - "time": 0.1, - "log_message": full_message, - "progress_message": full_message - } + """创建一个动作日志""" + full_message = f"{prefix} {emoji} {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"{prefix} {message}" - - return { - "action_name": "wait", - "action_kwargs": { - "time": 0.1, - "log_message": safe_message, - "progress_message": safe_message - } - } \ No newline at end of file + } diff --git a/unilabos/compile/utils/resource_helper.py b/unilabos/compile/utils/resource_helper.py new file mode 100644 index 00000000..343081e5 --- /dev/null +++ b/unilabos/compile/utils/resource_helper.py @@ -0,0 +1,172 @@ +""" +资源实例兼容层 + +提供 ensure_resource_instance() 将 dict / ResourceDictInstance 统一转为 +ResourceDictInstance，使编译器可以渐进式迁移到强类型资源。 +""" + +from typing import Any, Dict, Optional, Union + +from unilabos.resources.resource_tracker import ResourceDictInstance + + +def ensure_resource_instance( + resource: Union[Dict[str, Any], ResourceDictInstance, None], +) -> Optional[ResourceDictInstance]: + """将 dict 或 ResourceDictInstance 统一转为 ResourceDictInstance + + 编译器入口统一调用此函数，即可同时兼容旧 dict 传参和新 ResourceDictInstance 传参。 + + Args: + resource: 资源数据，可以是 plain dict、ResourceDictInstance 或 None + + Returns: + ResourceDictInstance 或 None（当输入为 None 时） + """ + if resource is None: + return None + if isinstance(resource, ResourceDictInstance): + return resource + if isinstance(resource, dict): + return ResourceDictInstance.get_resource_instance_from_dict(resource) + raise TypeError(f"不支持的资源类型: {type(resource)}, 期望 dict 或 ResourceDictInstance") + + +def resource_to_dict(resource: Union[Dict[str, Any], ResourceDictInstance]) -> Dict[str, Any]: + """将 ResourceDictInstance 或 dict 统一转为 plain dict + + 用于需要 dict 操作的场景（如 children dict 操作）。 + + Args: + resource: ResourceDictInstance 或 dict + + Returns: + plain dict + """ + if isinstance(resource, dict): + return resource + if isinstance(resource, ResourceDictInstance): + return resource.get_plr_nested_dict() + raise TypeError(f"不支持的资源类型: {type(resource)}") + + +def get_resource_id(resource: Union[str, Dict[str, Any], ResourceDictInstance]) -> str: + """从资源对象中提取 ID + + Args: + resource: 字符串 ID、dict 或 ResourceDictInstance + + Returns: + 资源 ID 字符串 + """ + if isinstance(resource, str): + return resource + if isinstance(resource, ResourceDictInstance): + return resource.res_content.id + if isinstance(resource, dict): + if "id" in resource: + return resource["id"] + # 兼容 {station_id: {...}} 格式 + first_val = next(iter(resource.values()), {}) + if isinstance(first_val, dict): + return first_val.get("id", "") + return "" + raise TypeError(f"不支持的资源类型: {type(resource)}") + + +def get_resource_data(resource: Union[str, Dict[str, Any], ResourceDictInstance]) -> Dict[str, Any]: + """从资源对象中提取 data 字段 + + Args: + resource: 字符串、dict 或 ResourceDictInstance + + Returns: + data 字典 + """ + if isinstance(resource, str): + return {} + if isinstance(resource, ResourceDictInstance): + return dict(resource.res_content.data) + if isinstance(resource, dict): + return resource.get("data", {}) + return {} + + +def get_resource_display_info(resource: Union[str, Dict[str, Any], ResourceDictInstance]) -> str: + """获取资源的显示信息（用于日志） + + Args: + resource: 字符串 ID、dict 或 ResourceDictInstance + + Returns: + 显示信息字符串 + """ + if isinstance(resource, str): + return resource + if isinstance(resource, ResourceDictInstance): + res = resource.res_content + return f"{res.id} ({res.name})" if res.name and res.name != res.id else res.id + if isinstance(resource, dict): + res_id = resource.get("id", "unknown") + res_name = resource.get("name", "") + if res_name and res_name != res_id: + return f"{res_id} ({res_name})" + return res_id + return str(resource) + + +def get_resource_liquid_volume(resource: Union[Dict[str, Any], ResourceDictInstance]) -> float: + """从资源中获取液体体积 + + Args: + resource: dict 或 ResourceDictInstance + + Returns: + 液体总体积 (mL) + """ + data = get_resource_data(resource) + liquids = data.get("liquid", []) + if isinstance(liquids, list): + return sum(l.get("volume", 0.0) for l in liquids if isinstance(l, dict)) + return 0.0 + + +def update_vessel_volume(vessel, G, new_volume: float, description: str = "") -> None: + """ + 更新容器体积（同时更新vessel字典和图节点） + + Args: + vessel: 容器字典或 ResourceDictInstance + G: 网络图 (nx.DiGraph) + new_volume: 新体积 (mL) + description: 更新描述（用于日志） + """ + import logging + logger = logging.getLogger("unilabos.compile") + + vessel_id = get_resource_id(vessel) + + if description: + logger.info(f"[RESOURCE] 更新容器体积 - {description}") + + # 更新 vessel 字典中的体积 + if isinstance(vessel, dict): + if "data" not in vessel: + vessel["data"] = {} + lv = vessel["data"].get("liquid_volume") + if isinstance(lv, list) and len(lv) > 0: + vessel["data"]["liquid_volume"][0] = new_volume + else: + vessel["data"]["liquid_volume"] = new_volume + + # 同时更新图中的容器数据 + if vessel_id and vessel_id in G.nodes(): + if "data" not in G.nodes[vessel_id]: + G.nodes[vessel_id]["data"] = {} + node_lv = G.nodes[vessel_id]["data"].get("liquid_volume") + if isinstance(node_lv, list) and len(node_lv) > 0: + G.nodes[vessel_id]["data"]["liquid_volume"][0] = new_volume + else: + G.nodes[vessel_id]["data"]["liquid_volume"] = new_volume + + logger.info(f"[RESOURCE] 容器 '{vessel_id}' 体积已更新为: {new_volume:.2f}mL") diff --git a/unilabos/compile/utils/unit_parser.py b/unilabos/compile/utils/unit_parser.py index 19a867bd..a4fe8cb9 100644 --- a/unilabos/compile/utils/unit_parser.py +++ b/unilabos/compile/utils/unit_parser.py @@ -184,6 +184,42 @@ def parse_time_input(time_input: Union[str, float]) -> float: return time_sec + +def parse_temperature_input(temp_input: Union[str, float], default_temp: float = 25.0) -> float: + """ + 解析温度输入，支持字符串和数值 + + Args: + temp_input: 温度输入（如 "256 °C", "reflux", 45.0） + default_temp: 默认温度 + + Returns: + float: 温度（°C） + """ + if not temp_input: + return default_temp + + if isinstance(temp_input, (int, float)): + return float(temp_input) + + temp_str = str(temp_input).lower().strip() + + # 特殊温度关键词 + special_temps = { + "room temperature": 25.0, "reflux": 78.0, "ice bath": 0.0, + "boiling": 100.0, "hot": 60.0, "warm": 40.0, "cold": 10.0, + } + if temp_str in special_temps: + return special_temps[temp_str] + + # 正则解析（如 "256 °C", "45°C", "45"） + match = re.search(r'(\d+(?:\.\d+)?)\s*°?[cf]?', temp_str) + if match: + return float(match.group(1)) + + debug_print(f"无法解析温度: '{temp_str}'，使用默认值: {default_temp}°C") + return default_temp + # 测试函数 def test_unit_parser(): """测试单位解析功能""" diff --git a/unilabos/compile/utils/vessel_parser.py b/unilabos/compile/utils/vessel_parser.py index a7bf673b..c0210298 100644 --- a/unilabos/compile/utils/vessel_parser.py +++ b/unilabos/compile/utils/vessel_parser.py @@ -1,27 +1,23 @@ import networkx as nx from .logger_util import debug_print +from .resource_helper import get_resource_id, get_resource_data def get_vessel(vessel): """ 统一处理vessel参数，返回vessel_id和vessel_data。 + 支持 dict、str、ResourceDictInstance。 Args: - vessel: 可以是一个字典或字符串，表示vessel的ID或数据。 + vessel: 可以是一个字典、字符串或 ResourceDictInstance，表示vessel的ID或数据。 Returns: tuple: 包含vessel_id和vessel_data。 """ - if isinstance(vessel, dict): - if "id" not in vessel: - vessel_id = list(vessel.values())[0].get("id", "") - else: - vessel_id = vessel.get("id", "") - vessel_data = vessel.get("data", {}) - else: - vessel_id = str(vessel) - vessel_data = {} + # 统一使用 resource_helper 处理 + vessel_id = get_resource_id(vessel) + vessel_data = get_resource_data(vessel) return vessel_id, vessel_data @@ -278,4 +274,31 @@ def find_solid_dispenser(G: nx.DiGraph) -> str: return node debug_print(f"❌ 未找到固体加样器") - return "" \ No newline at end of file + return "" + + +def find_connected_heatchill(G: nx.DiGraph, vessel: str) -> str: + """查找与指定容器相连的加热/冷却设备""" + heatchill_nodes = [] + for node in G.nodes(): + node_data = G.nodes[node] + node_class = node_data.get('class', '') or '' + node_name = node.lower() + if ('heatchill' in node_class.lower() or 'virtual_heatchill' in node_class + or 'heater' in node_name or 'heat' in node_name): + heatchill_nodes.append(node) + + # 检查连接 + if vessel and heatchill_nodes: + for hc in heatchill_nodes: + if G.has_edge(hc, vessel) or G.has_edge(vessel, hc): + debug_print(f"加热设备 '{hc}' 与容器 '{vessel}' 相连") + return hc + + # 使用第一个可用设备 + if heatchill_nodes: + debug_print(f"使用第一个加热设备: {heatchill_nodes[0]}") + return heatchill_nodes[0] + + debug_print("未找到加热设备，使用默认设备") + return "heatchill_1" \ No newline at end of file