89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
# components/device_components.py
|
||
"""
|
||
设备组件
|
||
"""
|
||
from typing import Dict, Any, Optional
|
||
import time
|
||
from core.component import Component, ComponentFactory
|
||
from core.exceptions import ComponentError
|
||
|
||
@ComponentFactory.register("wait_modbus_value")
|
||
class WaitModbusValueComponent(Component):
|
||
"""通用等待Modbus值(Name)组件"""
|
||
|
||
def execute(self) -> Dict[str, Any]:
|
||
# 获取参数
|
||
target_value = self.resolve_param("target_value")
|
||
name = self.resolve_param("name")
|
||
address = self.resolve_param("address")
|
||
description = self.resolve_param("description", "")
|
||
timeout = self.resolve_param("timeout", 30000) # 默认30秒超时
|
||
|
||
# 验证必要参数
|
||
self.validate_required_params(["target_value", "name"])
|
||
|
||
# 实际实现中，这里应该轮询Modbus设备直到值匹配或超时
|
||
# 这里简化为返回模拟数据
|
||
start_time = time.time() * 1000
|
||
current_value = None
|
||
success = False
|
||
|
||
# 模拟等待过程
|
||
while (time.time() * 1000 - start_time) < timeout:
|
||
# 模拟读取当前值
|
||
current_value = target_value # 假设值已经匹配
|
||
|
||
if current_value == target_value:
|
||
success = True
|
||
break
|
||
|
||
# 等待一段时间再次检查
|
||
time.sleep(0.1)
|
||
|
||
if not success:
|
||
result = {
|
||
"success": False,
|
||
"timeout": True,
|
||
"name": name,
|
||
"address": address,
|
||
"targetValue": target_value,
|
||
"currentValue": current_value
|
||
}
|
||
else:
|
||
result = {
|
||
"success": True,
|
||
"name": name,
|
||
"address": address,
|
||
"value": current_value
|
||
}
|
||
|
||
# 存储结果
|
||
self.store_result(result)
|
||
return result
|
||
|
||
@ComponentFactory.register("write_modbus_value")
|
||
class WriteModbusValueComponent(Component):
|
||
"""通用写入Modbus值(Name)组件"""
|
||
|
||
def execute(self) -> Dict[str, Any]:
|
||
# 获取参数
|
||
write_value = self.resolve_param("write_value")
|
||
name = self.resolve_param("name")
|
||
address = self.resolve_param("address")
|
||
description = self.resolve_param("description", "")
|
||
|
||
# 验证必要参数
|
||
self.validate_required_params(["write_value", "name"])
|
||
|
||
# 实际实现中，这里应该调用Modbus设备API写入值
|
||
# 这里简化为返回模拟数据
|
||
result = {
|
||
"success": True,
|
||
"name": name,
|
||
"address": address,
|
||
"value": write_value
|
||
}
|
||
|
||
# 存储结果
|
||
self.store_result(result)
|
||
return result |