Files
Uni-Lab-OS/unilabos/utils/pywinauto_util.py
Xuwznln 2a5ddd611d Upgrade to py 3.11.14; ROS2 Humble 0.7; unilabos 0.10.16
Workbench example, adjust log level, and ci check (#220)

* TestLatency Return Value Example & gitignore update

* Adjust log level & Add workbench virtual example & Add not action decorator & Add check_mode &

* Add CI Check

Fix/workstation yb revision (#217)

* Revert log change & update registry

* Revert opcua client & move electrolyte node

Workstation yb merge dev ready 260113 (#216)

* feat(bioyond): 添加计算实验设计功能,支持化合物配比和滴定比例参数

* feat(bioyond): 添加测量小瓶功能,支持基本参数配置

* feat(bioyond): 添加测量小瓶配置,支持新设备参数

* feat(bioyond): 更新仓库布局和尺寸,支持竖向排列的测量小瓶和试剂存放堆栈

* feat(bioyond): 优化任务创建流程,确保无论成功与否都清理任务队列以避免重复累积

* feat(bioyond): 添加设置反应器温度功能,支持温度范围和异常处理

* feat(bioyond): 调整反应器位置配置,统一坐标格式

* feat(bioyond): 添加调度器启动功能,支持任务队列执行并处理异常

* feat(bioyond): 优化调度器启动功能,添加异常处理并更新相关配置

* feat(opcua): 增强节点ID解析兼容性和数据类型处理

改进节点ID解析逻辑以支持多种格式,包括字符串和数字标识符
添加数据类型转换处理,确保写入值时类型匹配
优化错误提示信息,便于调试节点连接问题

* feat(registry): 新增后处理站的设备配置文件

添加后处理站的YAML配置文件,包含动作映射、状态类型和设备描述

* 添加调度器启动功能,合并物料参数配置,优化物料参数处理逻辑

* 添加从 Bioyond 系统自动同步工作流序列的功能,并更新相关配置

* fix:兼容 BioyondReactionStation 中 workflow_sequence 被重写为 property

* fix:同步工作流序列

* feat: remove commented workflow synchronization from `reaction_station.py`.

* 添加时间约束功能及相关配置

* fix:自动更新物料缓存功能,添加物料时更新缓存并在删除时移除缓存项

* fix:在添加物料时处理字符串和字典返回值,确保正确更新缓存

* fix:更新奔曜错误处理报送为物料变更报送,调整日志记录和响应消息

* feat:添加实验报告简化功能,去除冗余信息并保留关键信息

* feat: 添加任务状态事件发布功能,监控并报告任务运行、超时、完成和错误状态

* fix: 修复添加物料时数据格式错误

* Refactor bioyond_dispensing_station and reaction_station_bioyond YAML configurations

- Removed redundant action value mappings from bioyond_dispensing_station.
- Updated goal properties in bioyond_dispensing_station to use enums for target_stack and other parameters.
- Changed data types for end_point and start_point in reaction_station_bioyond to use string enums (Start, End).
- Simplified descriptions and updated measurement units from μL to mL where applicable.
- Removed unused commands from reaction_station_bioyond to streamline the configuration.

* fix:Change the material unit from μL to mL

* fix:refresh_material_cache

* feat: 动态获取工作流步骤ID,优化工作流配置

* feat: 添加清空服务端所有非核心工作流功能

* fix:修复Bottle类的序列化和反序列化方法

* feat:增强材料缓存更新逻辑,支持处理返回数据中的详细信息

* Add debug log

* feat(workstation): update bioyond config migration and coin cell material search logic

- Migrate bioyond_cell config to JSON structure and remove global variable dependencies
- Implement material search confirmation dialog auto-handling
- Add documentation: 20260113_物料搜寻确认弹窗自动处理功能.md and 20260113_配置迁移修改总结.md

* Refactor module paths for Bioyond devices in YAML configuration files

- Updated the module path for BioyondDispensingStation in bioyond_dispensing_station.yaml to reflect the new directory structure.
- Updated the module path for BioyondReactionStation and BioyondReactor in reaction_station_bioyond.yaml to align with the revised organization of the codebase.

* fix: WareHouse 的不可哈希类型错误,优化父节点去重逻辑

* refactor: Move config from module to instance initialization

* fix: 修正 reaction_station 目录名拼写错误

* feat: Integrate material search logic and cleanup deprecated files

- Update coin_cell_assembly.py with material search dialog handling
- Update YB_warehouses.py with latest warehouse configurations
- Remove outdated documentation and test data files

* Refactor: Use instance attributes for action names and workflow step IDs

* refactor: Split tipbox storage into left and right warehouses

* refactor: Merge tipbox storage left and right into single warehouse

---------

Co-authored-by: ZiWei <131428629+ZiWei09@users.noreply.github.com>
Co-authored-by: Andy6M <xieqiming1132@qq.com>

fix: WareHouse 的不可哈希类型错误,优化父节点去重逻辑

fix parent_uuid fetch when bind_parent_id == node_name

物料更新也是用父节点进行报送

Add None conversion for tube rack etc.

Add set_liquid example.

Add create_resource and test_resource example.

Add restart.
Temp allow action message.

Add no_update_feedback option.

Create session_id by edge.

bump version to 0.10.15

temp cancel update req
2026-01-27 15:21:55 +08:00

166 lines
6.2 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import psutil
import pywinauto
try:
from pywinauto_recorder import UIApplication
from pywinauto_recorder.player import UIPath, click, focus_on_application, exists, find, get_wrapper_path
except ImportError:
print("未安装pywinauto_recorder部分功能无法使用安装时注意enum")
pass
from pywinauto.controls.uiawrapper import UIAWrapper
from pywinauto.application import WindowSpecification
from pywinauto import findbestmatch
import sys
import codecs
import os
import locale
def connect_application(backend="uia", **kwargs):
app = pywinauto.Application(backend=backend)
app.connect(**kwargs)
top_window = app.top_window().wrapper_object()
native_window_handle = top_window.handle
return UIApplication(app, native_window_handle)
def get_ui_path_with_window_specification(obj):
return UIPath(get_wrapper_path(obj))
def get_process_pid_by_name(process_name: str, min_memory_mb: float = 0) -> tuple[bool, int]:
"""
通过进程名称和最小内存要求获取进程PID
Args:
process_name: 进程名称
min_memory_mb: 最小内存要求(MB)默认为0表示不检查内存
Returns:
tuple[bool, int]: (是否找到进程, 进程PID)
"""
process_found = False
process_pid = None
min_memory_bytes = min_memory_mb * 1024 * 1024 # 转换为字节
try:
for proc in psutil.process_iter(['name', 'pid', 'memory_info']):
try:
# 获取进程信息
proc_info = proc.info
if process_name in proc_info['name']:
# 如果设置了内存限制,则检查内存
if min_memory_mb > 0:
memory_info = proc_info.get('memory_info')
if memory_info and memory_info.rss > min_memory_bytes:
process_found = True
process_pid = proc_info['pid']
break
else:
# 不检查内存,直接返回找到的进程
process_found = True
process_pid = proc_info['pid']
break
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
continue
except Exception as e:
print(f"获取进程信息时发生错误: {str(e)}")
return process_found, process_pid
def print_wrapper_identifiers(wrapper_object, depth=None, filename=None):
"""
打印控件及其子控件的标识信息
Args:
wrapper_object: UIAWrapper对象
depth: 打印的最大深度,None表示打印全部
filename: 输出文件名,None表示打印到控制台
"""
if depth is None:
depth = sys.maxsize
# 创建所有控件的列表(当前控件及其所有子代)
all_ctrls = [wrapper_object, ] + wrapper_object.descendants()
# 创建所有可见文本控件的列表
txt_ctrls = [ctrl for ctrl in all_ctrls if ctrl.can_be_label and ctrl.is_visible() and ctrl.window_text()]
# 构建唯一的控件名称字典
name_ctrl_id_map = findbestmatch.UniqueDict()
for index, ctrl in enumerate(all_ctrls):
ctrl_names = findbestmatch.get_control_names(ctrl, all_ctrls, txt_ctrls)
for name in ctrl_names:
name_ctrl_id_map[name] = index
# 反转映射关系(控件索引到名称列表)
ctrl_id_name_map = {}
for name, index in name_ctrl_id_map.items():
ctrl_id_name_map.setdefault(index, []).append(name)
def print_identifiers(ctrls, current_depth=1, log_func=print):
"""递归打印控件及其子代的标识信息"""
if len(ctrls) == 0 or current_depth > depth:
return
indent = (current_depth - 1) * u" | "
for ctrl in ctrls:
try:
ctrl_id = all_ctrls.index(ctrl)
except ValueError:
continue
ctrl_text = ctrl.window_text()
if ctrl_text:
# 将多行文本转换为单行
ctrl_text = ctrl_text.replace('\n', r'\n').replace('\r', r'\r')
output = indent + u'\n'
output += indent + u"{class_name} - '{text}' {rect}\n"\
"".format(class_name=ctrl.friendly_class_name(),
text=ctrl_text,
rect=ctrl.rectangle())
output += indent + u'{}'.format(ctrl_id_name_map[ctrl_id])
title = ctrl_text
class_name = ctrl.class_name()
auto_id = None
control_type = None
if hasattr(ctrl.element_info, 'automation_id'):
auto_id = ctrl.element_info.automation_id
if hasattr(ctrl.element_info, 'control_type'):
control_type = ctrl.element_info.control_type
if control_type:
class_name = None # 如果有control_type就不需要class_name
else:
control_type = None # 如果control_type为空,仍使用class_name
criteria_texts = []
recorder_texts = []
if title:
criteria_texts.append(u'title="{}"'.format(title))
recorder_texts.append(f"{title}")
if class_name:
criteria_texts.append(u'class_name="{}"'.format(class_name))
if auto_id:
criteria_texts.append(u'auto_id="{}"'.format(auto_id))
if control_type:
criteria_texts.append(u'control_type="{}"'.format(control_type))
recorder_texts.append(f"||{control_type}")
if title or class_name or auto_id:
output += u'\n' + indent + u'child_window(' + u', '.join(criteria_texts) + u')' + " / " + "".join(recorder_texts)
log_func(output)
print_identifiers(ctrl.children(), current_depth + 1, log_func)
if filename is None:
print("Control Identifiers:")
print_identifiers([wrapper_object, ])
else:
log_file = codecs.open(filename, "w", locale.getpreferredencoding())
def log_func(msg):
log_file.write(str(msg) + os.linesep)
log_func("Control Identifiers:")
print_identifiers([wrapper_object, ], log_func=log_func)
log_file.close()