open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2025-12-15 13:44:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

unit convert

Browse Source
This commit is contained in:
王俊杰
2025-04-23 14:53:29 +08:00
parent 0cd11fa46b
commit 2f69480f92
6 changed files with 39 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
unilabos/compile/clean_protocol.py
View File

@@ -6,7 +6,7 @@ def generate_clean_protocol(
G: nx.DiGraph,
vessel: str, # Vessel to clean.
solvent: str, # Solvent to clean vessel with.
volume: float = 25000.0, # Optional. Volume of solvent to clean vessel with.
volume: float = 25.0, # Optional. Volume of solvent to clean vessel with.
temp: float = 25, # Optional. Temperature to heat vessel to while cleaning.
repeats: int = 1, # Optional. Number of cleaning cycles to perform.
) -> list[dict]:
@@ -27,7 +27,7 @@ def generate_clean_protocol(
from_vessel = f"flask_{solvent}"
waste_vessel = f"waste_workup"
transfer_flowrate = flowrate = 2500.0
transfer_flowrate = flowrate = 2.5
# 生成泵操作的动作序列
for i in range(repeats):

4
unilabos/compile/evaporate_protocol.py
View File

@@ -24,8 +24,8 @@ def generate_evaporate_protocol(
# 生成泵操作的动作序列
pump_action_sequence = []
reactor_volume = 500000.0
transfer_flowrate = flowrate = 2500.0
reactor_volume = 500.0
transfer_flowrate = flowrate = 2.5
# 开启冷凝器
pump_action_sequence.append({

16
unilabos/compile/pump_protocol.py
View File

@@ -7,7 +7,7 @@ def generate_pump_protocol(
from_vessel: str,
to_vessel: str,
volume: float,
flowrate: float = 500.0,
flowrate: float = 0.5,
transfer_flowrate: float = 0,
) -> list[dict]:
"""
@@ -141,11 +141,11 @@ def generate_pump_protocol_with_rinsing(
time: float = 0,
viscous: bool = False,
rinsing_solvent: str = "air",
rinsing_volume: float = 5000.0,
rinsing_volume: float = 5.0,
rinsing_repeats: int = 2,
solid: bool = False,
flowrate: float = 2500.0,
transfer_flowrate: float = 500.0,
flowrate: float = 2.5,
transfer_flowrate: float = 0.5,
) -> list[dict]:
"""
Generates a pump protocol for transferring a specified volume between vessels, including rinsing steps with a chosen solvent. This function constructs a sequence of pump actions based on the provided parameters and the shortest path in a directed graph.
@@ -159,11 +159,11 @@ def generate_pump_protocol_with_rinsing(
time (float, optional): Time over which to perform the transfer (default is 0).
viscous (bool, optional): Indicates if the fluid is viscous (default is False).
rinsing_solvent (str, optional): The solvent to use for rinsing (default is "air").
rinsing_volume (float, optional): The volume of rinsing solvent to use (default is 5000.0).
rinsing_volume (float, optional): The volume of rinsing solvent to use (default is 5.0).
rinsing_repeats (int, optional): The number of times to repeat rinsing (default is 2).
solid (bool, optional): Indicates if the transfer involves a solid (default is False).
flowrate (float, optional): The flow rate for the transfer (default is 2500.0). 最终注入容器B时的流速
transfer_flowrate (float, optional): The flow rate for the transfer action (default is 500.0). 泵骨架中转移流速（若不指定，默认与注入流速相同）
flowrate (float, optional): The flow rate for the transfer (default is 2.5). 最终注入容器B时的流速
transfer_flowrate (float, optional): The flow rate for the transfer action (default is 0.5). 泵骨架中转移流速（若不指定，默认与注入流速相同）
Returns:
list[dict]: A sequence of pump actions to be executed for the transfer and rinsing process. 泵操作的动作序列.
@@ -172,7 +172,7 @@ def generate_pump_protocol_with_rinsing(
AssertionError: If the number of rinsing solvents does not match the number of rinsing repeats.
Examples:
pump_protocol = generate_pump_protocol_with_rinsing(G, "vessel_A", "vessel_B", 100.0, rinsing_solvent="water")
pump_protocol = generate_pump_protocol_with_rinsing(G, "vessel_A", "vessel_B", 0.1, rinsing_solvent="water")
"""
air_vessel = "flask_air"
waste_vessel = f"waste_workup"

26
unilabos/compile/separate_protocol.py
View File

@@ -11,7 +11,7 @@ def generate_separate_protocol(
to_vessel: str, # Vessel to send product phase to.
waste_phase_to_vessel: str, # Optional. Vessel to send waste phase to.
solvent: str, # Optional. Solvent to add to separation vessel after contents of from_vessel has been transferred to create two phases.
solvent_volume: float = 50000, # Optional. Volume of solvent to add.
solvent_volume: float = 50, # Optional. Volume of solvent to add (mL).
through: str = "", # Optional. Solid chemical to send product phase through on way to to_vessel, e.g. 'celite'.
repeats: int = 1, # Optional. Number of separations to perform.
stir_time: float = 30, # Optional. Time stir for after adding solvent, before separation of phases.
@@ -32,7 +32,7 @@ def generate_separate_protocol(
# 生成泵操作的动作序列
pump_action_sequence = []
reactor_volume = 500000.0
reactor_volume = 500.0
waste_vessel = waste_phase_to_vessel
# TODO：通过物料管理系统找到溶剂的容器
@@ -46,7 +46,7 @@ def generate_separate_protocol(
separator_controller = f"{separation_vessel}_controller"
separation_vessel_bottom = f"flask_{separation_vessel}"
transfer_flowrate = flowrate = 2500.0
transfer_flowrate = flowrate = 2.5
if from_vessel != separation_vessel:
pump_action_sequence.append(
@@ -140,8 +140,8 @@ def generate_separate_protocol(
"action_kwargs": {
"from_vessel": separation_vessel_bottom,
"to_vessel": to_vessel,
"volume": 250000.0,
"time": 250000.0 / flowrate,
"volume": 250.0,
"time": 250.0 / flowrate,
# "transfer_flowrate": transfer_flowrate,
}
}
@@ -164,8 +164,8 @@ def generate_separate_protocol(
"action_kwargs": {
"from_vessel": separation_vessel_bottom,
"to_vessel": waste_vessel,
"volume": 250000.0,
"time": 250000.0 / flowrate,
"volume": 250.0,
"time": 250.0 / flowrate,
# "transfer_flowrate": transfer_flowrate,
}
}
@@ -179,8 +179,8 @@ def generate_separate_protocol(
"action_kwargs": {
"from_vessel": separation_vessel_bottom,
"to_vessel": waste_vessel,
"volume": 250000.0,
"time": 250000.0 / flowrate,
"volume": 250.0,
"time": 250.0 / flowrate,
# "transfer_flowrate": transfer_flowrate,
}
}
@@ -203,8 +203,8 @@ def generate_separate_protocol(
"action_kwargs": {
"from_vessel": separation_vessel_bottom,
"to_vessel": to_vessel,
"volume": 250000.0,
"time": 250000.0 / flowrate,
"volume": 250.0,
"time": 250.0 / flowrate,
# "transfer_flowrate": transfer_flowrate,
}
}
@@ -221,8 +221,8 @@ def generate_separate_protocol(
"action_kwargs": {
"from_vessel": to_vessel,
"to_vessel": separation_vessel,
"volume": 250000.0,
"time": 250000.0 / flowrate,
"volume": 250.0,
"time": 250.0 / flowrate,
# "transfer_flowrate": transfer_flowrate,
}
}

12
unilabos/ros/nodes/base_device_node.py
View File

@@ -342,9 +342,15 @@ class BaseROS2DeviceNode(Node, Generic[T]):
else:
return getattr(self.driver_instance, attr_name)
except AttributeError as ex:
self.lab_logger().error(
f"publish error, {str(type(self.driver_instance))[8:-2]} has no attribute '{attr_name}'"
)
if ex.args[0].startswith(f"AttributeError: '{self.driver_instance.__class__.__name__}'"):
self.lab_logger().error(
f"publish error, {str(type(self.driver_instance))[8:-2]} has no attribute '{attr_name}'"
)
else:
self.lab_logger().error(
f"publish error, when {str(type(self.driver_instance))[8:-2]} getting attribute '{attr_name}'"
)
self.lab_logger().error(traceback.format_exc())
self._property_publishers[attr_name] = PropertyPublisher(
self, attr_name, get_device_attr, msg_type, initial_period, self._print_publish

10
unilabos/ros/nodes/presets/serial_node.py
View File

@@ -47,7 +47,7 @@ class ROS2SerialNode(BaseROS2DeviceNode):
self.lab_logger().info(f"【ROS2SerialNode.__init__】创建串口写入服务: serialwrite")
def send_command(self, command: str):
self.lab_logger().info(f"【ROS2SerialNode.send_command】发送命令: {command}")
# self.lab_logger().debug(f"【ROS2SerialNode.send_command】发送命令: {command}")
with self._query_lock:
if self._closing:
self.lab_logger().error(f"【ROS2SerialNode.send_command】设备正在关闭，无法发送命令")
@@ -59,23 +59,23 @@ class ROS2SerialNode(BaseROS2DeviceNode):
response = self.hardware_interface.write(full_command_data)
# time.sleep(0.05)
output = self._receive(self.hardware_interface.read_until(b"\n"))
self.lab_logger().info(f"【ROS2SerialNode.send_command】接收响应: {output}")
# self.lab_logger().debug(f"【ROS2SerialNode.send_command】接收响应: {output}")
return output
def read_data(self):
self.lab_logger().debug(f"【ROS2SerialNode.read_data】读取数据")
# self.lab_logger().debug(f"【ROS2SerialNode.read_data】读取数据")
with self._query_lock:
if self._closing:
self.lab_logger().error(f"【ROS2SerialNode.read_data】设备正在关闭，无法读取数据")
raise RuntimeError
data = self.hardware_interface.read_until(b"\n")
result = self._receive(data)
self.lab_logger().debug(f"【ROS2SerialNode.read_data】读取到数据: {result}")
# self.lab_logger().debug(f"【ROS2SerialNode.read_data】读取到数据: {result}")
return result
def _receive(self, data: bytes):
ascii_string = "".join(chr(byte) for byte in data)
self.lab_logger().debug(f"【ROS2SerialNode._receive】接收数据: {ascii_string}")
# self.lab_logger().debug(f"【ROS2SerialNode._receive】接收数据: {ascii_string}")
return ascii_string
def handle_serial_request(self, request, response):