279 lines
12 KiB
Python
279 lines
12 KiB
Python
import time
|
||
import os
|
||
import random
|
||
import string
|
||
from framework.core.base_page import BasePage
|
||
from framework.core.logger import get_logger
|
||
|
||
logger = get_logger("QuantizationPage")
|
||
|
||
class QuantizationPage(BasePage):
|
||
"""量化工具页面 POM"""
|
||
|
||
MENU_TEXT = "量化工具"
|
||
|
||
def navigate_to(self):
|
||
"""进入量化工具页面"""
|
||
logger.info("正在切换到量化工具页面...")
|
||
self.smart_click(self.MENU_TEXT)
|
||
try:
|
||
self.page.wait_for_selector(".p-button:has-text('保存'), .p-tabmenu", timeout=10000)
|
||
except:
|
||
pass
|
||
time.sleep(2)
|
||
|
||
def click_create_task(self):
|
||
"""点击创建任务按钮"""
|
||
logger.info("正在点击创建任务按钮...")
|
||
self.smart_click("创建任务")
|
||
time.sleep(2)
|
||
|
||
def input_task_name(self, task_name="AutoQuantization"):
|
||
"""选中任务名称清空，输入随机任务名称"""
|
||
logger.info(f"📝 准备输入任务名称: {task_name}")
|
||
|
||
# 1. 精准定位：找到包含该 label 的直接父容器，从而限定在这个特定输入组内
|
||
input_box = self.page.locator("label").filter(has_text="任务名称").locator("..").locator("input").first
|
||
input_box.wait_for(state="visible", timeout=15000)
|
||
|
||
# 2. 强制清空
|
||
input_box.click()
|
||
# 识别系统
|
||
modifier = "Meta" if "mac" in str(self.page.context.browser.browser_type).lower() else "Control"
|
||
self.page.keyboard.press(f"{modifier}+A")
|
||
self.page.keyboard.press("Backspace")
|
||
|
||
# 3. 填入新值
|
||
input_box.fill(task_name)
|
||
time.sleep(1)
|
||
|
||
|
||
def navigate_to_quantization_task_config(self):
|
||
"""点击编译模型转换模式"""
|
||
logger.info("正在切换到量化任务配置页面...")
|
||
self.smart_click("编译模型转换模式")
|
||
try:
|
||
self.page.wait_for_selector(".p-button:has-text('保存'), .p-tabmenu", timeout=10000)
|
||
except:
|
||
pass
|
||
time.sleep(2)
|
||
|
||
"""点击工具版本"""
|
||
def click_tool_version(self):
|
||
logger.info("正在点击工具版本...")
|
||
self.smart_click("工具版本")
|
||
time.sleep(2)
|
||
|
||
def select_mode_fast_eval(self):
|
||
"""选择快速性能评测模式"""
|
||
logger.info("下拉选择: 快速性能评测模式")
|
||
# 修正: label 包在外部容器中，p-select 内部并没有该文本，需通过容器层级定位
|
||
container = self.page.locator("div:has(label:has-text('编译模型转换模式'))").first
|
||
dropdown = container.locator(".p-select").first
|
||
dropdown.click()
|
||
dropdown = container.locator(".p-select").first
|
||
dropdown.click()
|
||
|
||
# 弹出的选项列表中点击
|
||
option = self.page.locator("li.p-select-option:has-text('快速性能评测模式')")
|
||
option.click()
|
||
time.sleep(1)
|
||
|
||
def select_image(self, image_name="dcloud_ai_toolchain_ubuntu_22_s100_cpu-2"):
|
||
"""选择镜像流程"""
|
||
logger.info(f"点击选择镜像按钮，拉起弹窗...")
|
||
self.page.locator("button.p-button-link:has-text('选择镜像')").click()
|
||
|
||
# 在弹窗中定位镜像项
|
||
logger.info(f"📍 在弹窗中寻找镜像: {image_name}")
|
||
image_row = self.page.locator(f"div:has-text('{image_name}')").last
|
||
image_row.scroll_into_view_if_needed()
|
||
image_row.click()
|
||
|
||
# 点击弹窗内部的确定 (限制定位器在 p-dialog 容器内)
|
||
logger.info("⏳ 点击[镜像弹窗]确认按钮...")
|
||
dialog_confirm = self.page.locator(".p-dialog:has-text('群组镜像')").locator("button:has-text('确定')").last
|
||
dialog_confirm.click()
|
||
time.sleep(2)
|
||
|
||
def input_march_param(self, value="nash-e"):
|
||
"""模型参数 march 输入"""
|
||
logger.info(f"📝 模型参数输入 march: {value}")
|
||
# 精准定位：找到特定的 label，回到直接父容器，再找 input，防止全局 flex 匹配
|
||
march_input = self.page.locator("label").filter(has_text="march").locator("..").locator("input").first
|
||
march_input.scroll_into_view_if_needed()
|
||
|
||
# 强制清空再输入
|
||
march_input.click()
|
||
modifier = "Meta" if "mac" in str(self.page.context.browser.browser_type).lower() else "Control"
|
||
self.page.keyboard.press(f"{modifier}+A")
|
||
self.page.keyboard.press("Backspace")
|
||
|
||
march_input.fill(value)
|
||
time.sleep(1)
|
||
|
||
def click_final_submit(self):
|
||
"""点击页面最终确认按钮"""
|
||
logger.info("💾 点击量化任务最终[确认]按钮")
|
||
# 往往在页面正下方，不在弹窗内
|
||
confirm_btn = self.page.locator("button:has-text('确定')").last
|
||
confirm_btn.click()
|
||
logger.info("🎉 量化任务已成功提交")
|
||
time.sleep(2)
|
||
|
||
def upload_model_file(self, folder_name="model", file_name="result.onnx"):
|
||
"""点击模型文件框，选择具体模型文件"""
|
||
logger.info("📂 点击[模型文件]区域，拉起路径选择...")
|
||
# 定位虚线框
|
||
self.page.locator("div.border-dashed:has-text('点击上传模型')").click()
|
||
time.sleep(1)
|
||
|
||
# 在“选择文件”弹窗内操作
|
||
file_picker = self.page.locator(".p-dialog:has-text('选择文件')")
|
||
|
||
# 1. 点击左侧文件夹
|
||
logger.info(f"📍 进入文件夹: {folder_name}")
|
||
file_picker.locator(f"span.text-sm:has-text('{folder_name}')").click()
|
||
time.sleep(1)
|
||
|
||
# 2. 点击右侧文件列表中的目标项
|
||
logger.info(f"📄 选择具体文件: {file_name}")
|
||
file_picker.locator(f"span:has-text('{file_name}')").click()
|
||
time.sleep(1)
|
||
|
||
# 3. 点击弹窗底部的确定
|
||
logger.info("⏳ 点击[选择文件]弹窗确认按钮...")
|
||
file_picker.locator("button:has-text('确定')").last.click()
|
||
logger.info("✅ 模型文件选取成功")
|
||
time.sleep(1)
|
||
|
||
def open_and_close_logs(self, task_name):
|
||
"""等待任务进入运行中状态后，点击日志打开新窗口，等待加载后关闭"""
|
||
import re
|
||
logger.info(f"📄 准备查看任务 [{task_name}] 的日志...")
|
||
row = self.page.locator("tr").filter(has_text=task_name).first
|
||
|
||
# 1. 拦截队列：必须等到状态跳变出[排队中]
|
||
logger.info(f"⏳ 正在监听排队队列，等待任务步入 [运行中] 状态...")
|
||
state_loc = row.locator("span, div").filter(has_text=re.compile(r"运行中|已完成|失败")).first
|
||
state_loc.wait_for(state="visible", timeout=180000) # 最多等待3分钟排队
|
||
|
||
curr_state = state_loc.text_content()
|
||
if "运行中" not in curr_state:
|
||
logger.warning(f"⚠️ 状态跃迁过快或异常，当前状态已达 [{curr_state}]，无需再去查看动态运行日志。")
|
||
return
|
||
|
||
logger.info(f"✅ 状态刷新确认: [运行中]！正在拉起全屏日志流面板...")
|
||
|
||
# 2. 监听新页面的弹出
|
||
with self.page.context.expect_page() as new_page_info:
|
||
row.locator("button:has-text('日志')").click()
|
||
|
||
new_page = new_page_info.value
|
||
new_page.wait_for_load_state("domcontentloaded")
|
||
logger.info("✅ 已切入日志新窗口，等待底层视图渲染...")
|
||
new_page.locator("body").wait_for(state="visible", timeout=10000)
|
||
time.sleep(2) # 给足够的流式打印时间预览
|
||
new_page.close()
|
||
logger.info("✅ 日志窗口已平滑关闭，切回主视图。测试流继续下放。")
|
||
|
||
def stop_task(self, task_name):
|
||
"""停止任务并等待状态变为已停止"""
|
||
logger.info(f"⏹ 准备停止任务: {task_name}")
|
||
row = self.page.locator("tr").filter(has_text=task_name).first
|
||
|
||
# 等待停止按钮可点击
|
||
stop_btn = row.locator("button:has-text('停止')")
|
||
stop_btn.click()
|
||
|
||
logger.info("⏳ 等待任务状态变为 [已停止]...")
|
||
# 等待出现 已停止 的标签
|
||
row.locator("span, div").filter(has_text="已停止").first.wait_for(state="visible", timeout=60000)
|
||
logger.info("✅ 任务停止成功")
|
||
|
||
def delete_task(self, task_name):
|
||
"""删除已停止的任务"""
|
||
logger.info(f"🗑 正在删除任务: {task_name}")
|
||
row = self.page.locator("tr").filter(has_text=task_name).first
|
||
row.locator("button:has-text('删除')").click()
|
||
|
||
# 点击气泡确认
|
||
logger.info("点击确认删除弹窗...")
|
||
confirm_btn = self.page.locator(".p-confirm-popup-accept-button, button:has-text('确定')").last
|
||
confirm_btn.wait_for(state="visible", timeout=5000)
|
||
confirm_btn.click()
|
||
|
||
# 校验行是否消失
|
||
row.wait_for(state="hidden", timeout=10000)
|
||
logger.info(f"✅ 任务 [{task_name}] 已从列表中移除")
|
||
|
||
def wait_for_task_finished(self, task_name, timeout_mins=5):
|
||
"""轮询等待任务达到终态 (已完成 / 失败)"""
|
||
import re
|
||
logger.info(f"⏳ 正在监听任务 [{task_name}] 执行进度 (上限 {timeout_mins} 分钟)...")
|
||
row = self.page.locator("tr").filter(has_text=task_name).first
|
||
|
||
try:
|
||
# 同时匹配"已完成"或"失败"任一状态
|
||
end_state = row.locator("span, div").filter(has_text=re.compile(r"已完成|失败")).first
|
||
end_state.wait_for(state="visible", timeout=timeout_mins * 60000)
|
||
|
||
state_text = end_state.text_content()
|
||
if "失败" in state_text:
|
||
logger.error(f"❌ 任务执行异常: 监控到状态变为 [失败]")
|
||
raise Exception("后端量化任务处理失败")
|
||
else:
|
||
logger.info(f"🎉 业务通过: 任务执行成功结束，当前处于 [{state_text}] 状态")
|
||
time.sleep(2)
|
||
except Exception as e:
|
||
logger.error(f"❌ 监听任务进度时发生中断或超时: {e}")
|
||
raise e
|
||
|
||
def open_and_close_report(self, task_name):
|
||
"""点击报告打开新窗口，等待页面完全加载后关闭"""
|
||
logger.info(f"📊 准备查看任务 [{task_name}] 的检查报告...")
|
||
row = self.page.locator("tr").filter(has_text=task_name).first
|
||
|
||
# 拦截新标签页弹出
|
||
with self.page.context.expect_page() as new_page_info:
|
||
row.locator("button:has-text('报告')").click()
|
||
|
||
new_page = new_page_info.value
|
||
new_page.wait_for_load_state("domcontentloaded")
|
||
new_page.locator("body").wait_for(state="visible", timeout=15000)
|
||
logger.info("✅ 已切入[报告]新窗口，基础DOM渲染就绪...")
|
||
time.sleep(3) # 提供安全的静态视觉缓冲期保证页面完全加载
|
||
new_page.close()
|
||
logger.info("✅ 报告页面窗口已关闭，切回主视图。")
|
||
|
||
def download_all_results(self, task_name):
|
||
"""点击下载拉起弹窗，下载全部结果并等待下载任务完成后结束"""
|
||
logger.info(f"📥 准备下载任务 [{task_name}] 的全部输出结果...")
|
||
row = self.page.locator("tr").filter(has_text=task_name).first
|
||
|
||
# 1. 点击当前行下载动作
|
||
row.locator("button:has-text('下载')").click()
|
||
|
||
# 2. 接管全局下载弹窗
|
||
dialog = self.page.locator(".p-dialog").filter(has_text="下载全部结果")
|
||
dialog.wait_for(state="visible", timeout=5000)
|
||
|
||
# 3. 阻塞拦截底层下载事件
|
||
logger.info("⏳ 点击[下载全部结果]，安全挂起等待文件流写入完毕...")
|
||
dl_btn = dialog.locator("button:has-text('下载全部结果')")
|
||
with self.page.expect_download(timeout=60000) as download_info:
|
||
dl_btn.click()
|
||
|
||
download = download_info.value
|
||
# 主动调用 path() 即可在当前主线程强阻塞，直至磁盘下载动作完全落盘成功
|
||
dl_path = download.path()
|
||
logger.info(f"🎉 文件下载成功落盘！临时归档路径: {dl_path}")
|
||
time.sleep(2)
|
||
|
||
# 4. 收尾清理弹窗 UI
|
||
close_btn = dialog.locator("button.p-dialog-header-close").first
|
||
if close_btn.is_visible():
|
||
close_btn.click()
|
||
time.sleep(1)
|
||
|
||
logger.info("✅ 下载流程顺畅完结。") |