mirror of
https://github.com/dptech-corp/Uni-Lab-OS.git
synced 2026-02-05 14:05:12 +00:00
修复了添加protocol前缀导致的不能启动的bug
This commit is contained in:
KCFeng425
@@ -1,20 +1,34 @@
|
||||
import time
|
||||
from typing import Union
|
||||
from typing import Union, Dict, Optional
|
||||
|
||||
|
||||
class VirtualMultiwayValve:
|
||||
"""
|
||||
虚拟八通阀门 - 可将一个输入切换到8个输出中的任意一个
|
||||
虚拟九通阀门 - 0号位连接transfer pump，1-8号位连接其他设备
|
||||
"""
|
||||
def __init__(self, port: str = "VIRTUAL", positions: int = 8):
|
||||
self.port = port
|
||||
self.max_positions = positions
|
||||
self.max_positions = positions # 1-8号位
|
||||
self.total_positions = positions + 1 # 0-8号位，共9个位置
|
||||
|
||||
# 状态属性
|
||||
self._status = "Idle"
|
||||
self._valve_state = "Ready"
|
||||
self._current_position = 1
|
||||
self._target_position = 1
|
||||
self._current_position = 0 # 默认在0号位（transfer pump位置）
|
||||
self._target_position = 0
|
||||
|
||||
# 位置映射说明
|
||||
self.position_map = {
|
||||
0: "transfer_pump", # 0号位连接转移泵
|
||||
1: "port_1", # 1号位
|
||||
2: "port_2", # 2号位
|
||||
3: "port_3", # 3号位
|
||||
4: "port_4", # 4号位
|
||||
5: "port_5", # 5号位
|
||||
6: "port_6", # 6号位
|
||||
7: "port_7", # 7号位
|
||||
8: "port_8" # 8号位
|
||||
}
|
||||
|
||||
@property
|
||||
def status(self) -> str:
|
||||
@@ -36,12 +50,18 @@ class VirtualMultiwayValve:
|
||||
"""获取当前阀门位置"""
|
||||
return self._current_position
|
||||
|
||||
def get_current_port(self) -> str:
|
||||
"""获取当前连接的端口名称"""
|
||||
return self.position_map.get(self._current_position, "unknown")
|
||||
|
||||
def set_position(self, command: Union[int, str]):
|
||||
"""
|
||||
设置阀门位置 - 兼容 SendCmd 类型
|
||||
设置阀门位置 - 支持0-8位置
|
||||
|
||||
Args:
|
||||
command: 目标位置 (1-8) 或位置字符串
|
||||
command: 目标位置 (0-8) 或位置字符串
|
||||
0: transfer pump位置
|
||||
1-8: 其他设备位置
|
||||
"""
|
||||
try:
|
||||
# 如果是字符串形式的位置，先转换为数字
|
||||
@@ -50,8 +70,8 @@ class VirtualMultiwayValve:
|
||||
else:
|
||||
pos = int(command)
|
||||
|
||||
if pos < 1 or pos > self.max_positions:
|
||||
raise ValueError(f"Position must be between 1 and {self.max_positions}")
|
||||
if pos < 0 or pos > self.max_positions:
|
||||
raise ValueError(f"Position must be between 0 and {self.max_positions}")
|
||||
|
||||
self._status = "Busy"
|
||||
self._valve_state = "Moving"
|
||||
@@ -65,28 +85,44 @@ class VirtualMultiwayValve:
|
||||
self._status = "Idle"
|
||||
self._valve_state = "Ready"
|
||||
|
||||
return f"Position set to {pos}"
|
||||
current_port = self.get_current_port()
|
||||
return f"Position set to {pos} ({current_port})"
|
||||
|
||||
except ValueError as e:
|
||||
self._status = "Error"
|
||||
self._valve_state = "Error"
|
||||
return f"Error: {str(e)}"
|
||||
|
||||
def set_to_pump_position(self):
|
||||
"""切换到transfer pump位置（0号位）"""
|
||||
return self.set_position(0)
|
||||
|
||||
def set_to_port(self, port_number: int):
|
||||
"""
|
||||
切换到指定端口位置
|
||||
|
||||
Args:
|
||||
port_number: 端口号 (1-8)
|
||||
"""
|
||||
if port_number < 1 or port_number > self.max_positions:
|
||||
raise ValueError(f"Port number must be between 1 and {self.max_positions}")
|
||||
return self.set_position(port_number)
|
||||
|
||||
def open(self):
|
||||
"""打开阀门 - 对于多通阀门，相当于设置到位置1"""
|
||||
return self.set_position(1)
|
||||
"""打开阀门 - 设置到transfer pump位置（0号位）"""
|
||||
return self.set_to_pump_position()
|
||||
|
||||
def close(self):
|
||||
"""关闭阀门 - 对于多通阀门，相当于设置到关闭位置"""
|
||||
"""关闭阀门 - 对于多通阀门，设置到一个"关闭"状态"""
|
||||
self._status = "Busy"
|
||||
self._valve_state = "Closing"
|
||||
time.sleep(0.5)
|
||||
|
||||
self._current_position = 0 # 0表示关闭状态
|
||||
# 可以选择保持当前位置或设置特殊关闭状态
|
||||
self._status = "Idle"
|
||||
self._valve_state = "Closed"
|
||||
|
||||
return "Valve closed"
|
||||
return f"Valve closed at position {self._current_position}"
|
||||
|
||||
def get_valve_position(self) -> int:
|
||||
"""获取阀门位置 - 兼容性方法"""
|
||||
@@ -96,21 +132,90 @@ class VirtualMultiwayValve:
|
||||
"""检查是否在指定位置"""
|
||||
return self._current_position == position
|
||||
|
||||
def is_at_pump_position(self) -> bool:
|
||||
"""检查是否在transfer pump位置"""
|
||||
return self._current_position == 0
|
||||
|
||||
def is_at_port(self, port_number: int) -> bool:
|
||||
"""检查是否在指定端口位置"""
|
||||
return self._current_position == port_number
|
||||
|
||||
def get_available_positions(self) -> list:
|
||||
"""获取可用位置列表"""
|
||||
return list(range(1, self.max_positions + 1))
|
||||
return list(range(0, self.max_positions + 1))
|
||||
|
||||
def get_available_ports(self) -> Dict[int, str]:
|
||||
"""获取可用端口映射"""
|
||||
return self.position_map.copy()
|
||||
|
||||
def reset(self):
|
||||
"""重置阀门到初始位置"""
|
||||
return self.set_position(1)
|
||||
"""重置阀门到transfer pump位置（0号位）"""
|
||||
return self.set_position(0)
|
||||
|
||||
def switch_between_pump_and_port(self, port_number: int):
|
||||
"""
|
||||
在transfer pump位置和指定端口之间切换
|
||||
|
||||
Args:
|
||||
port_number: 目标端口号 (1-8)
|
||||
"""
|
||||
if self._current_position == 0:
|
||||
# 当前在pump位置，切换到指定端口
|
||||
return self.set_to_port(port_number)
|
||||
else:
|
||||
# 当前在某个端口，切换到pump位置
|
||||
return self.set_to_pump_position()
|
||||
|
||||
def get_flow_path(self) -> str:
|
||||
"""获取当前流路路径描述"""
|
||||
current_port = self.get_current_port()
|
||||
if self._current_position == 0:
|
||||
return f"Transfer pump connected (position {self._current_position})"
|
||||
else:
|
||||
return f"Port {self._current_position} connected ({current_port})"
|
||||
|
||||
def get_info(self) -> dict:
|
||||
"""获取阀门信息"""
|
||||
"""获取阀门详细信息"""
|
||||
return {
|
||||
"port": self.port,
|
||||
"max_positions": self.max_positions,
|
||||
"total_positions": self.total_positions,
|
||||
"current_position": self._current_position,
|
||||
"current_port": self.get_current_port(),
|
||||
"target_position": self._target_position,
|
||||
"status": self._status,
|
||||
"valve_state": self._valve_state
|
||||
}
|
||||
"valve_state": self._valve_state,
|
||||
"flow_path": self.get_flow_path(),
|
||||
"position_map": self.position_map
|
||||
}
|
||||
|
||||
def __str__(self):
|
||||
return f"VirtualMultiwayValve(Position: {self._current_position}/{self.max_positions}, Port: {self.get_current_port()}, Status: {self._status})"
|
||||
|
||||
|
||||
# 使用示例
|
||||
if __name__ == "__main__":
|
||||
valve = VirtualMultiwayValve()
|
||||
|
||||
print("=== 虚拟九通阀门测试 ===")
|
||||
print(f"初始状态: {valve}")
|
||||
print(f"当前流路: {valve.get_flow_path()}")
|
||||
|
||||
# 切换到试剂瓶1（1号位）
|
||||
print(f"\n切换到1号位: {valve.set_position(1)}")
|
||||
print(f"当前状态: {valve}")
|
||||
|
||||
# 切换到transfer pump位置（0号位）
|
||||
print(f"\n切换到pump位置: {valve.set_to_pump_position()}")
|
||||
print(f"当前状态: {valve}")
|
||||
|
||||
# 切换到试剂瓶2（2号位）
|
||||
print(f"\n切换到2号位: {valve.set_to_port(2)}")
|
||||
print(f"当前状态: {valve}")
|
||||
|
||||
# 显示所有可用位置
|
||||
print(f"\n可用位置: {valve.get_available_positions()}")
|
||||
print(f"端口映射: {valve.get_available_ports()}")
|
||||
|
||||
# 获取详细信息
|
||||
print(f"\n详细信息: {valve.get_info()}")
|
||||
0
unilabos/devices/virtual/virtual_rotavap.py
Normal file
0
unilabos/devices/virtual/virtual_rotavap.py
Normal file
@@ -1,149 +1,285 @@
|
||||
import asyncio
|
||||
import time
|
||||
from enum import Enum
|
||||
from typing import Union, Optional
|
||||
import logging
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
class VirtualTransferPump:
|
||||
"""Virtual pump device specifically for Transfer protocol"""
|
||||
|
||||
class VirtualPumpMode(Enum):
|
||||
Normal = 0
|
||||
AccuratePos = 1
|
||||
AccuratePosVel = 2
|
||||
|
||||
|
||||
class VirtualPump:
|
||||
"""虚拟泵类 - 模拟泵的基本功能，无需实际硬件"""
|
||||
|
||||
def __init__(self, device_id: str = None, config: Dict[str, Any] = None, **kwargs):
|
||||
# 处理可能的不同调用方式
|
||||
if device_id is None and 'id' in kwargs:
|
||||
device_id = kwargs.pop('id')
|
||||
if config is None and 'config' in kwargs:
|
||||
config = kwargs.pop('config')
|
||||
def __init__(self, device_id: str = None, max_volume: float = 25.0, mode: VirtualPumpMode = VirtualPumpMode.Normal):
|
||||
self.device_id = device_id or "virtual_pump"
|
||||
self.max_volume = max_volume
|
||||
self.mode = mode
|
||||
|
||||
# 设置默认值
|
||||
self.device_id = device_id or "unknown_transfer_pump"
|
||||
self.config = config or {}
|
||||
# 状态变量
|
||||
self._status = "Idle"
|
||||
self._position = 0.0 # 当前柱塞位置 (ml)
|
||||
self._max_velocity = 5.0 # 默认最大速度 (ml/s)
|
||||
self._current_volume = 0.0 # 当前注射器中的体积
|
||||
|
||||
self.logger = logging.getLogger(f"VirtualTransferPump.{self.device_id}")
|
||||
self.data = {}
|
||||
self.logger = logging.getLogger(f"VirtualPump.{self.device_id}")
|
||||
|
||||
# 添加调试信息
|
||||
print(f"=== VirtualTransferPump {self.device_id} is being created! ===")
|
||||
print(f"=== Config: {self.config} ===")
|
||||
print(f"=== Kwargs: {kwargs} ===")
|
||||
|
||||
# 从config或kwargs中获取配置参数
|
||||
self.port = self.config.get('port') or kwargs.get('port', 'VIRTUAL')
|
||||
self._max_volume = self.config.get('max_volume') or kwargs.get('max_volume', 50.0)
|
||||
self._transfer_rate = self.config.get('transfer_rate') or kwargs.get('transfer_rate', 5.0)
|
||||
self._current_volume = 0.0
|
||||
self.is_running = False
|
||||
|
||||
async def initialize(self) -> bool:
|
||||
"""Initialize virtual transfer pump"""
|
||||
print(f"=== VirtualTransferPump {self.device_id} initialize() called! ===")
|
||||
self.logger.info(f"Initializing virtual transfer pump {self.device_id}")
|
||||
self.data.update({
|
||||
"status": "Idle",
|
||||
"current_volume": 0.0,
|
||||
"max_volume": self._max_volume,
|
||||
"transfer_rate": self._transfer_rate,
|
||||
"from_vessel": "",
|
||||
"to_vessel": "",
|
||||
"progress": 0.0,
|
||||
"transferred_volume": 0.0,
|
||||
"current_status": "Ready"
|
||||
})
|
||||
"""初始化虚拟泵"""
|
||||
self.logger.info(f"Initializing virtual pump {self.device_id}")
|
||||
self._status = "Idle"
|
||||
self._position = 0.0
|
||||
self._current_volume = 0.0
|
||||
return True
|
||||
|
||||
async def cleanup(self) -> bool:
|
||||
"""Cleanup virtual transfer pump"""
|
||||
self.logger.info(f"Cleaning up virtual transfer pump {self.device_id}")
|
||||
"""清理虚拟泵"""
|
||||
self.logger.info(f"Cleaning up virtual pump {self.device_id}")
|
||||
self._status = "Idle"
|
||||
return True
|
||||
|
||||
async def transfer(self, from_vessel: str, to_vessel: str, volume: float,
|
||||
amount: str = "", time: float = 0, viscous: bool = False,
|
||||
rinsing_solvent: str = "", rinsing_volume: float = 0.0,
|
||||
rinsing_repeats: int = 0, solid: bool = False) -> bool:
|
||||
"""Execute liquid transfer - matches Transfer action"""
|
||||
self.logger.info(f"Transfer: {volume}mL from {from_vessel} to {to_vessel}")
|
||||
|
||||
# 计算转移时间
|
||||
if time > 0:
|
||||
transfer_time = time
|
||||
else:
|
||||
# 如果是粘性液体，降低转移速率
|
||||
rate = self._transfer_rate * 0.5 if viscous else self._transfer_rate
|
||||
transfer_time = volume / rate
|
||||
|
||||
self.data.update({
|
||||
"status": "Running",
|
||||
"from_vessel": from_vessel,
|
||||
"to_vessel": to_vessel,
|
||||
"current_status": "Transferring",
|
||||
"progress": 0.0,
|
||||
"transferred_volume": 0.0
|
||||
})
|
||||
|
||||
# 模拟转移过程
|
||||
steps = 10
|
||||
step_time = transfer_time / steps
|
||||
step_volume = volume / steps
|
||||
|
||||
for i in range(steps):
|
||||
await asyncio.sleep(step_time)
|
||||
progress = (i + 1) / steps * 100
|
||||
transferred = (i + 1) * step_volume
|
||||
|
||||
self.data.update({
|
||||
"progress": progress,
|
||||
"transferred_volume": transferred,
|
||||
"current_status": f"Transferring {progress:.1f}%"
|
||||
})
|
||||
|
||||
self.logger.info(f"Transfer progress: {progress:.1f}% ({transferred:.1f}/{volume}mL)")
|
||||
|
||||
# 如果需要冲洗
|
||||
if rinsing_solvent and rinsing_volume > 0 and rinsing_repeats > 0:
|
||||
self.data["current_status"] = "Rinsing"
|
||||
for repeat in range(rinsing_repeats):
|
||||
self.logger.info(f"Rinsing cycle {repeat + 1}/{rinsing_repeats} with {rinsing_solvent}")
|
||||
await asyncio.sleep(1) # 模拟冲洗时间
|
||||
|
||||
self.data.update({
|
||||
"status": "Idle",
|
||||
"current_status": "Transfer completed",
|
||||
"progress": 100.0,
|
||||
"transferred_volume": volume
|
||||
})
|
||||
|
||||
return True
|
||||
|
||||
# 添加所有在virtual_device.yaml中定义的状态属性
|
||||
# 基本属性
|
||||
@property
|
||||
def status(self) -> str:
|
||||
return self.data.get("status", "Unknown")
|
||||
return self._status
|
||||
|
||||
@property
|
||||
def position(self) -> float:
|
||||
"""当前柱塞位置 (ml)"""
|
||||
return self._position
|
||||
|
||||
@property
|
||||
def current_volume(self) -> float:
|
||||
return self.data.get("current_volume", 0.0)
|
||||
"""当前注射器中的体积 (ml)"""
|
||||
return self._current_volume
|
||||
|
||||
@property
|
||||
def max_volume(self) -> float:
|
||||
return self.data.get("max_volume", self._max_volume)
|
||||
def max_velocity(self) -> float:
|
||||
return self._max_velocity
|
||||
|
||||
@property
|
||||
def transfer_rate(self) -> float:
|
||||
return self.data.get("transfer_rate", self._transfer_rate)
|
||||
def set_max_velocity(self, velocity: float):
|
||||
"""设置最大速度 (ml/s)"""
|
||||
self._max_velocity = max(0.1, min(50.0, velocity)) # 限制在合理范围内
|
||||
self.logger.info(f"Set max velocity to {self._max_velocity} ml/s")
|
||||
|
||||
@property
|
||||
def from_vessel(self) -> str:
|
||||
return self.data.get("from_vessel", "")
|
||||
def get_status(self) -> str:
|
||||
"""获取泵状态"""
|
||||
return self._status
|
||||
|
||||
@property
|
||||
def to_vessel(self) -> str:
|
||||
return self.data.get("to_vessel", "")
|
||||
async def _simulate_operation(self, duration: float):
|
||||
"""模拟操作延时"""
|
||||
self._status = "Busy"
|
||||
await asyncio.sleep(duration)
|
||||
self._status = "Idle"
|
||||
|
||||
@property
|
||||
def progress(self) -> float:
|
||||
return self.data.get("progress", 0.0)
|
||||
def _calculate_duration(self, volume: float, velocity: float = None) -> float:
|
||||
"""计算操作持续时间"""
|
||||
if velocity is None:
|
||||
velocity = self._max_velocity
|
||||
return abs(volume) / velocity
|
||||
|
||||
@property
|
||||
def transferred_volume(self) -> float:
|
||||
return self.data.get("transferred_volume", 0.0)
|
||||
# 基本泵操作
|
||||
async def set_position(self, position: float, velocity: float = None):
|
||||
"""
|
||||
移动到绝对位置
|
||||
|
||||
Args:
|
||||
position (float): 目标位置 (ml)
|
||||
velocity (float): 移动速度 (ml/s)
|
||||
"""
|
||||
position = max(0, min(self.max_volume, position)) # 限制在有效范围内
|
||||
|
||||
volume_to_move = abs(position - self._position)
|
||||
duration = self._calculate_duration(volume_to_move, velocity)
|
||||
|
||||
self.logger.info(f"Moving to position {position} ml (current: {self._position} ml)")
|
||||
|
||||
# 模拟移动过程
|
||||
await self._simulate_operation(duration)
|
||||
|
||||
self._position = position
|
||||
self._current_volume = position # 假设位置等于体积
|
||||
|
||||
self.logger.info(f"Reached position {self._position} ml")
|
||||
|
||||
@property
|
||||
def current_status(self) -> str:
|
||||
return self.data.get("current_status", "Ready")
|
||||
async def pull_plunger(self, volume: float, velocity: float = None):
|
||||
"""
|
||||
拉取柱塞（吸液）
|
||||
|
||||
Args:
|
||||
volume (float): 要拉取的体积 (ml)
|
||||
velocity (float): 拉取速度 (ml/s)
|
||||
"""
|
||||
new_position = min(self.max_volume, self._position + volume)
|
||||
actual_volume = new_position - self._position
|
||||
|
||||
if actual_volume <= 0:
|
||||
self.logger.warning("Cannot pull - already at maximum volume")
|
||||
return
|
||||
|
||||
duration = self._calculate_duration(actual_volume, velocity)
|
||||
|
||||
self.logger.info(f"Pulling {actual_volume} ml (from {self._position} to {new_position})")
|
||||
|
||||
await self._simulate_operation(duration)
|
||||
|
||||
self._position = new_position
|
||||
self._current_volume = new_position
|
||||
|
||||
self.logger.info(f"Pulled {actual_volume} ml, current volume: {self._current_volume} ml")
|
||||
|
||||
async def push_plunger(self, volume: float, velocity: float = None):
|
||||
"""
|
||||
推出柱塞（排液）
|
||||
|
||||
Args:
|
||||
volume (float): 要推出的体积 (ml)
|
||||
velocity (float): 推出速度 (ml/s)
|
||||
"""
|
||||
new_position = max(0, self._position - volume)
|
||||
actual_volume = self._position - new_position
|
||||
|
||||
if actual_volume <= 0:
|
||||
self.logger.warning("Cannot push - already at minimum volume")
|
||||
return
|
||||
|
||||
duration = self._calculate_duration(actual_volume, velocity)
|
||||
|
||||
self.logger.info(f"Pushing {actual_volume} ml (from {self._position} to {new_position})")
|
||||
|
||||
await self._simulate_operation(duration)
|
||||
|
||||
self._position = new_position
|
||||
self._current_volume = new_position
|
||||
|
||||
self.logger.info(f"Pushed {actual_volume} ml, current volume: {self._current_volume} ml")
|
||||
|
||||
# 便捷操作方法
|
||||
async def aspirate(self, volume: float, velocity: float = None):
|
||||
"""
|
||||
吸液操作
|
||||
|
||||
Args:
|
||||
volume (float): 吸液体积 (ml)
|
||||
velocity (float): 吸液速度 (ml/s)
|
||||
"""
|
||||
await self.pull_plunger(volume, velocity)
|
||||
|
||||
async def dispense(self, volume: float, velocity: float = None):
|
||||
"""
|
||||
排液操作
|
||||
|
||||
Args:
|
||||
volume (float): 排液体积 (ml)
|
||||
velocity (float): 排液速度 (ml/s)
|
||||
"""
|
||||
await self.push_plunger(volume, velocity)
|
||||
|
||||
async def transfer(self, volume: float, aspirate_velocity: float = None, dispense_velocity: float = None):
|
||||
"""
|
||||
转移操作（先吸后排）
|
||||
|
||||
Args:
|
||||
volume (float): 转移体积 (ml)
|
||||
aspirate_velocity (float): 吸液速度 (ml/s)
|
||||
dispense_velocity (float): 排液速度 (ml/s)
|
||||
"""
|
||||
# 吸液
|
||||
await self.aspirate(volume, aspirate_velocity)
|
||||
|
||||
# 短暂停顿
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# 排液
|
||||
await self.dispense(volume, dispense_velocity)
|
||||
|
||||
async def empty_syringe(self, velocity: float = None):
|
||||
"""清空注射器"""
|
||||
await self.set_position(0, velocity)
|
||||
|
||||
async def fill_syringe(self, velocity: float = None):
|
||||
"""充满注射器"""
|
||||
await self.set_position(self.max_volume, velocity)
|
||||
|
||||
async def stop_operation(self):
|
||||
"""停止当前操作"""
|
||||
self._status = "Idle"
|
||||
self.logger.info("Operation stopped")
|
||||
|
||||
# 状态查询方法
|
||||
def get_position(self) -> float:
|
||||
"""获取当前位置"""
|
||||
return self._position
|
||||
|
||||
def get_current_volume(self) -> float:
|
||||
"""获取当前体积"""
|
||||
return self._current_volume
|
||||
|
||||
def get_remaining_capacity(self) -> float:
|
||||
"""获取剩余容量"""
|
||||
return self.max_volume - self._current_volume
|
||||
|
||||
def is_empty(self) -> bool:
|
||||
"""检查是否为空"""
|
||||
return self._current_volume <= 0.01 # 允许小量误差
|
||||
|
||||
def is_full(self) -> bool:
|
||||
"""检查是否已满"""
|
||||
return self._current_volume >= (self.max_volume - 0.01) # 允许小量误差
|
||||
|
||||
# 调试和状态信息
|
||||
def get_pump_info(self) -> dict:
|
||||
"""获取泵的详细信息"""
|
||||
return {
|
||||
"device_id": self.device_id,
|
||||
"status": self._status,
|
||||
"position": self._position,
|
||||
"current_volume": self._current_volume,
|
||||
"max_volume": self.max_volume,
|
||||
"max_velocity": self._max_velocity,
|
||||
"mode": self.mode.name,
|
||||
"is_empty": self.is_empty(),
|
||||
"is_full": self.is_full(),
|
||||
"remaining_capacity": self.get_remaining_capacity()
|
||||
}
|
||||
|
||||
def __str__(self):
|
||||
return f"VirtualPump({self.device_id}: {self._current_volume:.2f}/{self.max_volume} ml, {self._status})"
|
||||
|
||||
def __repr__(self):
|
||||
return self.__str__()
|
||||
|
||||
|
||||
# 使用示例
|
||||
async def demo():
|
||||
"""虚拟泵使用示例"""
|
||||
pump = VirtualPump("demo_pump", max_volume=50.0)
|
||||
|
||||
await pump.initialize()
|
||||
|
||||
print(f"Initial state: {pump}")
|
||||
|
||||
# 吸液测试
|
||||
await pump.aspirate(10.0, velocity=2.0)
|
||||
print(f"After aspirating 10ml: {pump}")
|
||||
|
||||
# 排液测试
|
||||
await pump.dispense(5.0, velocity=3.0)
|
||||
print(f"After dispensing 5ml: {pump}")
|
||||
|
||||
# 转移测试
|
||||
await pump.transfer(3.0)
|
||||
print(f"After transfer 3ml: {pump}")
|
||||
|
||||
# 清空测试
|
||||
await pump.empty_syringe()
|
||||
print(f"After emptying: {pump}")
|
||||
|
||||
print("\nPump info:", pump.get_pump_info())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(demo())
|
||||
@@ -89,7 +89,7 @@ mock_filter:
|
||||
target_volume: Float64
|
||||
action_value_mappings:
|
||||
filter:
|
||||
type: ProtocolFilter
|
||||
type: Filter
|
||||
goal:
|
||||
vessel: vessel
|
||||
filtrate_vessel: filtrate_vessel
|
||||
@@ -737,7 +737,7 @@ mock_stirrer_new:
|
||||
max_stir_speed: Float64
|
||||
action_value_mappings:
|
||||
start_stir:
|
||||
type: ProtocolStartStir
|
||||
type: StartStir
|
||||
goal:
|
||||
vessel: vessel
|
||||
stir_speed: stir_speed
|
||||
@@ -760,7 +760,7 @@ mock_stirrer_new:
|
||||
result:
|
||||
success: success
|
||||
stop_stir:
|
||||
type: ProtocolStopStir
|
||||
type: StopStir
|
||||
goal:
|
||||
vessel: vessel
|
||||
feedback:
|
||||
|
||||
@@ -76,7 +76,7 @@ virtual_pump:
|
||||
set_valve_position:
|
||||
type: FloatSingleInput
|
||||
goal:
|
||||
Int32: Int32
|
||||
float_in: valve_position
|
||||
feedback:
|
||||
status: status
|
||||
result:
|
||||
@@ -129,7 +129,7 @@ virtual_stirrer:
|
||||
result:
|
||||
success: success
|
||||
start_stir:
|
||||
type: ProtocolStartStir
|
||||
type: StartStir
|
||||
goal:
|
||||
vessel: vessel
|
||||
stir_speed: stir_speed
|
||||
@@ -139,7 +139,7 @@ virtual_stirrer:
|
||||
result:
|
||||
success: success
|
||||
stop_stir:
|
||||
type: ProtocolStopStir
|
||||
type: StopStir
|
||||
goal:
|
||||
vessel: vessel
|
||||
feedback:
|
||||
@@ -249,6 +249,13 @@ virtual_multiway_valve:
|
||||
data_source: executor
|
||||
data_key: fluid_port_7
|
||||
description: "八通阀门端口7，position=7时流体从此口流出"
|
||||
- handler_key: multiway-valve-port-8
|
||||
label: Port 8
|
||||
data_type: fluid
|
||||
io_type: source
|
||||
data_source: executor
|
||||
data_key: fluid_port_8
|
||||
description: "八通阀门端口8，position=8时流体从此口流出"
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
@@ -270,14 +277,16 @@ virtual_solenoid_valve:
|
||||
is_open: Bool
|
||||
action_value_mappings:
|
||||
open:
|
||||
type: EmptyIn
|
||||
goal: {}
|
||||
type: SendCmd
|
||||
goal:
|
||||
command: "open"
|
||||
feedback: {}
|
||||
result:
|
||||
success: success
|
||||
close:
|
||||
type: EmptyIn
|
||||
goal: {}
|
||||
type: SendCmd
|
||||
goal:
|
||||
command: "close"
|
||||
feedback: {}
|
||||
result:
|
||||
success: success
|
||||
@@ -329,7 +338,7 @@ virtual_centrifuge:
|
||||
time_remaining: Float64
|
||||
action_value_mappings:
|
||||
centrifuge:
|
||||
type: ProtocolCentrifuge
|
||||
type: Centrifuge
|
||||
goal:
|
||||
vessel: vessel
|
||||
speed: speed
|
||||
@@ -396,7 +405,7 @@ virtual_filter:
|
||||
message: String
|
||||
action_value_mappings:
|
||||
filter_sample:
|
||||
type: ProtocolFilter
|
||||
type: Filter
|
||||
goal:
|
||||
vessel: vessel
|
||||
filtrate_vessel: filtrate_vessel
|
||||
@@ -535,7 +544,7 @@ virtual_transfer_pump:
|
||||
current_status: String
|
||||
action_value_mappings:
|
||||
transfer:
|
||||
type: ProtocolTransfer
|
||||
type: Transfer
|
||||
goal:
|
||||
from_vessel: from_vessel
|
||||
to_vessel: to_vessel
|
||||
@@ -595,7 +604,7 @@ virtual_column:
|
||||
current_status: String
|
||||
action_value_mappings:
|
||||
run_column:
|
||||
type: ProtocolRunColumn
|
||||
type: RunColumn
|
||||
goal:
|
||||
from_vessel: from_vessel
|
||||
to_vessel: to_vessel
|
||||
|
||||
Reference in New Issue
Block a user