重构拆分设备连接/配置
This commit is contained in:
xinzhu.yin
90
app/config_io.py
Normal file
90
app/config_io.py
Normal file
@@ -0,0 +1,90 @@
|
||||
"""配置文件 I/O（Step 4 重构）。
|
||||
|
||||
从 pqAutomationApp.PQAutomationApp 中搬迁。每个函数第一行 `self = app`
|
||||
以保留原有 `self.xxx` 属性访问不变。
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
def get_config_path(app):
|
||||
"""获取配置文件的完整路径（兼容打包后的程序）"""
|
||||
self = app
|
||||
import os
|
||||
import sys
|
||||
|
||||
# 判断是否是打包后的程序
|
||||
if getattr(sys, "frozen", False):
|
||||
# 打包后：使用可执行文件所在目录
|
||||
base_path = os.path.dirname(sys.executable)
|
||||
else:
|
||||
# 开发环境：使用项目根目录（app/config_io.py 的上两级 = 工程根）
|
||||
base_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
# 构建配置文件路径
|
||||
config_dir = os.path.join(base_path, "settings")
|
||||
config_file = os.path.join(config_dir, "pq_config.json")
|
||||
|
||||
# 确保 settings 目录存在
|
||||
if not os.path.exists(config_dir):
|
||||
os.makedirs(config_dir)
|
||||
|
||||
return config_file
|
||||
|
||||
|
||||
def load_pq_config(app):
|
||||
"""加载PQ配置（兼容打包后的程序）"""
|
||||
self = app
|
||||
try:
|
||||
# ✅ 使用 self.config_file（已经是动态路径）
|
||||
if os.path.exists(self.config_file):
|
||||
with open(self.config_file, "r", encoding="utf-8") as f:
|
||||
config_dict = json.load(f)
|
||||
self.config.from_dict(config_dict)
|
||||
if hasattr(self, "log_gui"):
|
||||
self.log_gui.log("✓ 配置文件加载成功")
|
||||
else:
|
||||
if hasattr(self, "log_gui"):
|
||||
self.log_gui.log("⚠️ 配置文件不存在，使用默认配置")
|
||||
except Exception as e:
|
||||
if hasattr(self, "log_gui"):
|
||||
self.log_gui.log(f"⚠️ 加载配置文件失败: {str(e)}，使用默认配置")
|
||||
|
||||
|
||||
def save_pq_config(app):
|
||||
"""保存PQ配置（兼容打包后的程序）"""
|
||||
self = app
|
||||
try:
|
||||
# 确保目录存在
|
||||
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
|
||||
|
||||
# 保存配置
|
||||
self.config.save_to_file(self.config_file)
|
||||
except Exception as e:
|
||||
if hasattr(self, "log_gui"):
|
||||
self.log_gui.log(f"保存配置文件失败: {str(e)}")
|
||||
|
||||
|
||||
def clear_config_file(app):
|
||||
"""清理配置文件（兼容打包后的程序）"""
|
||||
self = app
|
||||
from tkinter import messagebox
|
||||
|
||||
config_file = self.get_config_path()
|
||||
|
||||
try:
|
||||
if os.path.exists(config_file):
|
||||
os.remove(config_file)
|
||||
self.config_cleared = True
|
||||
messagebox.showinfo("提示", "✓ 清理成功")
|
||||
self.log_gui.log("✓ 配置文件清理成功")
|
||||
else:
|
||||
messagebox.showinfo("提示", "配置文件不存在")
|
||||
self.log_gui.log("⚠️ 配置文件不存在")
|
||||
|
||||
except Exception as e:
|
||||
messagebox.showerror("错误", "❌ 清理失败")
|
||||
self.log_gui.log(f"❌ 配置文件清理失败: {str(e)}")
|
||||
|
||||
|
||||
0
app/device/__init__.py
Normal file
0
app/device/__init__.py
Normal file
210
app/device/connection.py
Normal file
210
app/device/connection.py
Normal file
@@ -0,0 +1,210 @@
|
||||
"""设备连接（UCD323 / CA410）相关逻辑（Step 4 重构）。
|
||||
|
||||
从 pqAutomationApp.PQAutomationApp 中搬迁。每个函数第一行 `self = app`
|
||||
以保留原有 `self.xxx` 属性访问不变。
|
||||
"""
|
||||
|
||||
import threading
|
||||
import time
|
||||
from tkinter import messagebox
|
||||
|
||||
from utils.caSerail import CASerail
|
||||
|
||||
def get_available_ucd_ports(app):
|
||||
"""获取可用的UCD端口列表"""
|
||||
self = app
|
||||
return self.ucd.search_device()
|
||||
|
||||
|
||||
def get_available_com_ports(app):
|
||||
"""获取可用的COM端口列表"""
|
||||
self = app
|
||||
try:
|
||||
import serial.tools.list_ports
|
||||
|
||||
ports = serial.tools.list_ports.comports()
|
||||
return [port.device for port in ports]
|
||||
except Exception as e:
|
||||
self.log_gui.log(f"获取COM端口列表出错: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def refresh_com_ports(app):
|
||||
"""刷新COM端口列表"""
|
||||
self = app
|
||||
available_ports = self.get_available_com_ports()
|
||||
available_list = self.get_available_ucd_ports()
|
||||
|
||||
# 更新UCD列表的下拉框选项
|
||||
ucd_list_current = self.ucd_list_var.get()
|
||||
if ucd_list_current not in available_list:
|
||||
self.ucd_list_var.set(available_list[0] if available_list else "")
|
||||
self.ucd_list_combo.config(values=available_list)
|
||||
|
||||
# 更新CA端口的下拉框选项
|
||||
ca_com_current = self.ca_com_var.get()
|
||||
if ca_com_current not in available_ports:
|
||||
self.ca_com_var.set(
|
||||
available_ports[1]
|
||||
if len(available_ports) > 1
|
||||
else (available_ports[0] if available_ports else "")
|
||||
)
|
||||
self.ca_com_combo.config(values=available_ports)
|
||||
|
||||
# 重置连接状态指示器为灰色
|
||||
if hasattr(self, "ucd_status_indicator"):
|
||||
self.ucd_status_indicator.config(bg="gray")
|
||||
if hasattr(self, "ca_status_indicator"):
|
||||
self.ca_status_indicator.config(bg="gray")
|
||||
|
||||
self.update_config()
|
||||
|
||||
|
||||
def check_com_connections(app):
|
||||
"""检测COM端口连接状态"""
|
||||
self = app
|
||||
# 禁用连接按钮，防止重复点击
|
||||
self.check_button.configure(state="disabled")
|
||||
self.refresh_button.configure(state="disabled")
|
||||
|
||||
# 更新状态栏
|
||||
self.status_var.set("正在检测连接...")
|
||||
self.root.update()
|
||||
|
||||
# 使用线程进行连接检测
|
||||
def check_connections():
|
||||
try:
|
||||
# 检测TV连接
|
||||
ucd_connected = self.check_port_connection(is_ucd=True)
|
||||
self.root.after(
|
||||
0,
|
||||
lambda: self.update_connection_indicator(
|
||||
self.ucd_status_indicator, ucd_connected
|
||||
),
|
||||
)
|
||||
|
||||
# 检测CA连接
|
||||
ca_connected = self.check_port_connection(is_ucd=False)
|
||||
self.root.after(
|
||||
0,
|
||||
lambda: self.update_connection_indicator(
|
||||
self.ca_status_indicator, ca_connected
|
||||
),
|
||||
)
|
||||
|
||||
# 更新状态栏
|
||||
self.root.after(0, lambda: self.status_var.set("连接检测完成"))
|
||||
|
||||
# 重新启用所有控件
|
||||
self.root.after(0, self.enable_com_widgets)
|
||||
except Exception as e:
|
||||
self.root.after(0, lambda: self.log_gui.log(f"连接检测出错: {e}"))
|
||||
self.root.after(0, self.enable_com_widgets)
|
||||
|
||||
# 启动线程
|
||||
threading.Thread(target=check_connections, daemon=True).start()
|
||||
|
||||
|
||||
def update_connection_indicator(app, indicator, connected):
|
||||
"""更新连接状态指示器颜色"""
|
||||
self = app
|
||||
if connected:
|
||||
indicator.config(bg="green")
|
||||
else:
|
||||
indicator.config(bg="red")
|
||||
|
||||
|
||||
def check_port_connection(app, is_ucd=True):
|
||||
"""检测指定端口是否可以连接"""
|
||||
self = app
|
||||
try:
|
||||
if is_ucd:
|
||||
if self.ucd.status:
|
||||
try:
|
||||
self.ucd.close()
|
||||
except:
|
||||
pass
|
||||
if not self.ucd.open(self.ucd_list_var.get()):
|
||||
self.log_gui.log(
|
||||
f"设备 {self.ucd_list_var.get()} 异常，UCD323连接失败"
|
||||
)
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
else:
|
||||
# 如果CA对象已存在，先关闭
|
||||
if self.ca is not None:
|
||||
try:
|
||||
self.ca.close()
|
||||
except:
|
||||
pass
|
||||
channel_value = self.ca_channel_var.get()
|
||||
str_channel = f"{int(channel_value):02d}"
|
||||
self.ca = CASerail()
|
||||
self.ca.open(self.config.device_config["ca_com"], 19200, 7, "E", 2)
|
||||
# data = self.ca.set_xyLv_Display()
|
||||
data = self.ca.set_all_Display()
|
||||
if data:
|
||||
data = self.ca.setSynchMode(3)
|
||||
data = self.ca.setMeasureSpeed(1)
|
||||
if True:
|
||||
time.sleep(0.5)
|
||||
data = self.ca.setZeroCalibration()
|
||||
channel_value = self.ca_channel_var.get()
|
||||
str_channel = f"{int(channel_value):02d}"
|
||||
data = self.ca.setChannel(str_channel)
|
||||
return True
|
||||
else:
|
||||
self.log_gui.log(
|
||||
f"端口 {self.config.device_config["ca_com"]} 异常，色温仪连接失败"
|
||||
)
|
||||
self.ca.close()
|
||||
self.ca = None
|
||||
return False
|
||||
except Exception as e:
|
||||
self.log_gui.log(f"端口连接失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def enable_com_widgets(app):
|
||||
"""重新启用所有控件"""
|
||||
self = app
|
||||
self.check_button.configure(state="normal")
|
||||
self.refresh_button.configure(state="normal")
|
||||
|
||||
|
||||
def disconnect_com_connections(app):
|
||||
"""断开所有串口连接"""
|
||||
self = app
|
||||
try:
|
||||
# 断开TV连接
|
||||
if self.ucd.status:
|
||||
try:
|
||||
self.ucd.close()
|
||||
except:
|
||||
pass
|
||||
finally:
|
||||
self.ucd.status = False
|
||||
self.log_gui.log("UCD连接已断开")
|
||||
|
||||
# 断开CA连接
|
||||
if self.ca is not None:
|
||||
try:
|
||||
self.ca.close()
|
||||
except:
|
||||
pass
|
||||
finally:
|
||||
self.ca = None
|
||||
self.log_gui.log("CA连接已断开")
|
||||
|
||||
# 重新启用相关控件
|
||||
self.enable_com_widgets()
|
||||
self.ucd_status_indicator.config(bg="gray")
|
||||
self.ca_status_indicator.config(bg="gray")
|
||||
self.status_var.set("串口连接已断开")
|
||||
|
||||
except Exception as e:
|
||||
self.log_gui.log(f"断开连接时发生错误: {str(e)}")
|
||||
messagebox.showerror("错误", f"断开连接失败: {str(e)}")
|
||||
|
||||
|
||||
228
app/tests/local_dimming.py
Normal file
228
app/tests/local_dimming.py
Normal file
@@ -0,0 +1,228 @@
|
||||
"""Local Dimming 测试逻辑（Step 4 重构）。
|
||||
|
||||
从 pqAutomationApp.PQAutomationApp 中搬迁。每个函数第一行 `self = app`
|
||||
以保留原有 `self.xxx` 属性访问不变。
|
||||
"""
|
||||
|
||||
import threading
|
||||
import tkinter as tk
|
||||
from tkinter import filedialog, messagebox
|
||||
|
||||
def start_local_dimming_test(app):
|
||||
"""开始 Local Dimming 测试"""
|
||||
self = app
|
||||
# 检查设备连接
|
||||
if not self.ca or not self.ucd.status:
|
||||
messagebox.showerror("错误", "请先连接 CA410 和 UCD323")
|
||||
return
|
||||
|
||||
# 禁用按钮
|
||||
self.ld_start_btn.config(state=tk.DISABLED)
|
||||
self.ld_stop_btn.config(state=tk.NORMAL)
|
||||
self.ld_save_btn.config(state=tk.DISABLED)
|
||||
|
||||
# 清空结果
|
||||
for item in self.ld_tree.get_children():
|
||||
self.ld_tree.delete(item)
|
||||
|
||||
# 获取配置
|
||||
wait_time = float(self.ld_wait_time_var.get())
|
||||
|
||||
# 在新线程中执行测试
|
||||
def run_test():
|
||||
from utils.local_dimming_test import LocalDimmingTest, LocalDimmingController
|
||||
|
||||
# 从设备当前 timing 获取分辨率
|
||||
ld_ctrl = LocalDimmingController(self.ucd)
|
||||
cur_w, cur_h = ld_ctrl.get_current_resolution()
|
||||
resolution = f"{cur_w}x{cur_h}"
|
||||
|
||||
ld_test = LocalDimmingTest(
|
||||
self.ucd,
|
||||
self.ca,
|
||||
log_callback=self.log_gui.log,
|
||||
)
|
||||
|
||||
ld_test.wait_time = wait_time
|
||||
|
||||
results = ld_test.run_test(resolution=resolution)
|
||||
|
||||
# 保存到实例变量
|
||||
self.ld_test_instance = ld_test
|
||||
self.ld_test_results = results
|
||||
|
||||
# 更新结果显示
|
||||
self.root.after(0, lambda: self.update_ld_results(results))
|
||||
|
||||
# 清理临时文件
|
||||
ld_test.cleanup()
|
||||
|
||||
# 恢复按钮状态
|
||||
self.root.after(0, lambda: self.ld_start_btn.config(state=tk.NORMAL))
|
||||
self.root.after(0, lambda: self.ld_stop_btn.config(state=tk.DISABLED))
|
||||
self.root.after(0, lambda: self.ld_save_btn.config(state=tk.NORMAL))
|
||||
|
||||
threading.Thread(target=run_test, daemon=True).start()
|
||||
|
||||
|
||||
def update_ld_results(app, results):
|
||||
"""更新 Local Dimming 结果显示"""
|
||||
self = app
|
||||
for percentage, x, y, lv, X, Y, Z in results:
|
||||
self.ld_tree.insert(
|
||||
"",
|
||||
tk.END,
|
||||
values=(f"{percentage}%", f"{lv:.2f}", f"{x:.4f}", f"{y:.4f}"),
|
||||
)
|
||||
|
||||
|
||||
def stop_local_dimming_test(app):
|
||||
"""停止 Local Dimming 测试"""
|
||||
self = app
|
||||
if hasattr(self, "ld_test_instance"):
|
||||
self.ld_test_instance.stop()
|
||||
|
||||
|
||||
def send_ld_window(app, percentage):
|
||||
"""发送指定百分比的窗口"""
|
||||
self = app
|
||||
if not self.ucd.status:
|
||||
messagebox.showwarning("警告", "请先连接 UCD323 设备")
|
||||
return
|
||||
|
||||
self.log_gui.log(f"🔆 发送 {percentage}% 窗口...")
|
||||
|
||||
# 记录当前百分比（用于测量）
|
||||
self.current_ld_percentage = percentage
|
||||
|
||||
def send():
|
||||
from utils.local_dimming_test import LocalDimmingController
|
||||
|
||||
ld_controller = LocalDimmingController(self.ucd)
|
||||
|
||||
# 从设备当前 timing 获取分辨率
|
||||
width, height = ld_controller.get_current_resolution()
|
||||
|
||||
# 生成并发送图片
|
||||
success = ld_controller.send_window_pattern_with_resolution(
|
||||
percentage, width, height
|
||||
)
|
||||
|
||||
if success:
|
||||
self.root.after(
|
||||
0, lambda: self.log_gui.log(f"✅ {percentage}% 窗口已发送")
|
||||
)
|
||||
else:
|
||||
self.root.after(
|
||||
0, lambda: self.log_gui.log(f"❌ {percentage}% 窗口发送失败")
|
||||
)
|
||||
|
||||
threading.Thread(target=send, daemon=True).start()
|
||||
|
||||
|
||||
def measure_ld_luminance(app):
|
||||
"""测量当前亮度"""
|
||||
self = app
|
||||
if not self.ca:
|
||||
messagebox.showwarning("警告", "请先连接 CA410 色度计")
|
||||
return
|
||||
|
||||
if self.current_ld_percentage is None:
|
||||
messagebox.showinfo("提示", "请先发送一个窗口图案")
|
||||
return
|
||||
|
||||
self.log_gui.log("📏 正在采集亮度...")
|
||||
|
||||
def measure():
|
||||
try:
|
||||
x, y, lv, X, Y, Z = self.ca.readAllDisplay()
|
||||
|
||||
if lv is not None:
|
||||
import datetime
|
||||
|
||||
timestamp = datetime.datetime.now().strftime("%H:%M:%S")
|
||||
|
||||
# 更新显示
|
||||
self.root.after(
|
||||
0,
|
||||
lambda: self.ld_result_label.config(
|
||||
text=f"亮度: {lv:.2f} cd/m² | x: {x:.4f} | y: {y:.4f}"
|
||||
),
|
||||
)
|
||||
|
||||
# 添加到表格
|
||||
self.root.after(
|
||||
0,
|
||||
lambda: self.ld_tree.insert(
|
||||
"",
|
||||
tk.END,
|
||||
values=(
|
||||
f"{self.current_ld_percentage}%",
|
||||
f"{lv:.2f}",
|
||||
f"{x:.4f}",
|
||||
f"{y:.4f}",
|
||||
timestamp,
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
self.root.after(
|
||||
0, lambda: self.log_gui.log(f"✅ 采集完成: {lv:.2f} cd/m²")
|
||||
)
|
||||
else:
|
||||
self.root.after(0, lambda: self.log_gui.log("❌ 采集失败"))
|
||||
|
||||
except Exception as e:
|
||||
self.root.after(0, lambda: self.log_gui.log(f"❌ 采集异常: {str(e)}"))
|
||||
|
||||
threading.Thread(target=measure, daemon=True).start()
|
||||
|
||||
|
||||
def clear_ld_records(app):
|
||||
"""清空测试记录"""
|
||||
self = app
|
||||
for item in self.ld_tree.get_children():
|
||||
self.ld_tree.delete(item)
|
||||
self.ld_result_label.config(text="亮度: -- cd/m² | x: -- | y: --")
|
||||
self.current_ld_percentage = None
|
||||
self.log_gui.log("🗑️ 测试记录已清空")
|
||||
|
||||
|
||||
def save_local_dimming_results(app):
|
||||
"""保存 Local Dimming 结果"""
|
||||
self = app
|
||||
from tkinter import filedialog
|
||||
import csv
|
||||
import datetime
|
||||
|
||||
if len(self.ld_tree.get_children()) == 0:
|
||||
messagebox.showinfo("提示", "没有可保存的数据")
|
||||
return
|
||||
|
||||
default_name = f"LocalDimming_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
|
||||
|
||||
save_path = filedialog.asksaveasfilename(
|
||||
title="保存测试结果",
|
||||
initialfile=default_name,
|
||||
defaultextension=".csv",
|
||||
filetypes=[("CSV 文件", "*.csv"), ("所有文件", "*.*")],
|
||||
)
|
||||
|
||||
if not save_path:
|
||||
return
|
||||
|
||||
try:
|
||||
with open(save_path, "w", newline="", encoding="utf-8-sig") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["窗口百分比", "亮度 (cd/m²)", "x", "y", "时间"])
|
||||
|
||||
for item in self.ld_tree.get_children():
|
||||
values = self.ld_tree.item(item, "values")
|
||||
writer.writerow(values)
|
||||
|
||||
self.log_gui.log(f"✓ 测试结果已保存: {save_path}")
|
||||
messagebox.showinfo("成功", f"测试结果已保存到:\n{save_path}")
|
||||
|
||||
except Exception as e:
|
||||
self.log_gui.log(f"❌ 保存失败: {str(e)}")
|
||||
messagebox.showerror("错误", f"保存失败: {str(e)}")
|
||||
@@ -65,6 +65,31 @@ from app.views.chart_frame import (
|
||||
on_chart_tab_changed as _cf_on_chart_tab_changed,
|
||||
update_chart_tabs_state as _cf_update_chart_tabs_state,
|
||||
)
|
||||
from app.config_io import (
|
||||
clear_config_file as _cfg_clear_config_file,
|
||||
get_config_path as _cfg_get_config_path,
|
||||
load_pq_config as _cfg_load_pq_config,
|
||||
save_pq_config as _cfg_save_pq_config,
|
||||
)
|
||||
from app.tests.local_dimming import (
|
||||
clear_ld_records as _ld_clear_ld_records,
|
||||
measure_ld_luminance as _ld_measure_ld_luminance,
|
||||
save_local_dimming_results as _ld_save_local_dimming_results,
|
||||
send_ld_window as _ld_send_ld_window,
|
||||
start_local_dimming_test as _ld_start_local_dimming_test,
|
||||
stop_local_dimming_test as _ld_stop_local_dimming_test,
|
||||
update_ld_results as _ld_update_ld_results,
|
||||
)
|
||||
from app.device.connection import (
|
||||
check_com_connections as _dev_check_com_connections,
|
||||
check_port_connection as _dev_check_port_connection,
|
||||
disconnect_com_connections as _dev_disconnect_com_connections,
|
||||
enable_com_widgets as _dev_enable_com_widgets,
|
||||
get_available_com_ports as _dev_get_available_com_ports,
|
||||
get_available_ucd_ports as _dev_get_available_ucd_ports,
|
||||
refresh_com_ports as _dev_refresh_com_ports,
|
||||
update_connection_indicator as _dev_update_connection_indicator,
|
||||
)
|
||||
|
||||
plt.rcParams["font.family"] = ["sans-serif"]
|
||||
plt.rcParams["font.sans-serif"] = ["Microsoft YaHei"]
|
||||
@@ -187,28 +212,8 @@ class PQAutomationApp:
|
||||
self.status_bar.pack(side=tk.BOTTOM, fill=tk.X)
|
||||
|
||||
def get_config_path(self):
|
||||
"""获取配置文件的完整路径（兼容打包后的程序）"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
# 判断是否是打包后的程序
|
||||
if getattr(sys, "frozen", False):
|
||||
# 打包后：使用可执行文件所在目录
|
||||
base_path = os.path.dirname(sys.executable)
|
||||
else:
|
||||
# 开发环境：使用脚本所在目录
|
||||
base_path = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
# 构建配置文件路径
|
||||
config_dir = os.path.join(base_path, "settings")
|
||||
config_file = os.path.join(config_dir, "pq_config.json")
|
||||
|
||||
# 确保 settings 目录存在
|
||||
if not os.path.exists(config_dir):
|
||||
os.makedirs(config_dir)
|
||||
|
||||
return config_file
|
||||
|
||||
"""转发到 app.config_io.get_config_path（Step 4 重构）"""
|
||||
return _cfg_get_config_path(self)
|
||||
def initialize_default_test_type(self):
|
||||
"""初始化默认测试类型（在所有控件创建完成后调用）"""
|
||||
try:
|
||||
@@ -1514,187 +1519,21 @@ class PQAutomationApp:
|
||||
ca_channel_combo.bind("<<ComboboxSelected>>", self.update_config)
|
||||
|
||||
def get_available_ucd_ports(self):
|
||||
"""获取可用的UCD端口列表"""
|
||||
return self.ucd.search_device()
|
||||
|
||||
return _dev_get_available_ucd_ports(self)
|
||||
def get_available_com_ports(self):
|
||||
"""获取可用的COM端口列表"""
|
||||
try:
|
||||
import serial.tools.list_ports
|
||||
|
||||
ports = serial.tools.list_ports.comports()
|
||||
return [port.device for port in ports]
|
||||
except Exception as e:
|
||||
self.log_gui.log(f"获取COM端口列表出错: {e}")
|
||||
return []
|
||||
|
||||
return _dev_get_available_com_ports(self)
|
||||
def refresh_com_ports(self):
|
||||
"""刷新COM端口列表"""
|
||||
available_ports = self.get_available_com_ports()
|
||||
available_list = self.get_available_ucd_ports()
|
||||
|
||||
# 更新UCD列表的下拉框选项
|
||||
ucd_list_current = self.ucd_list_var.get()
|
||||
if ucd_list_current not in available_list:
|
||||
self.ucd_list_var.set(available_list[0] if available_list else "")
|
||||
self.ucd_list_combo.config(values=available_list)
|
||||
|
||||
# 更新CA端口的下拉框选项
|
||||
ca_com_current = self.ca_com_var.get()
|
||||
if ca_com_current not in available_ports:
|
||||
self.ca_com_var.set(
|
||||
available_ports[1]
|
||||
if len(available_ports) > 1
|
||||
else (available_ports[0] if available_ports else "")
|
||||
)
|
||||
self.ca_com_combo.config(values=available_ports)
|
||||
|
||||
# 重置连接状态指示器为灰色
|
||||
if hasattr(self, "ucd_status_indicator"):
|
||||
self.ucd_status_indicator.config(bg="gray")
|
||||
if hasattr(self, "ca_status_indicator"):
|
||||
self.ca_status_indicator.config(bg="gray")
|
||||
|
||||
self.update_config()
|
||||
|
||||
return _dev_refresh_com_ports(self)
|
||||
def check_com_connections(self):
|
||||
"""检测COM端口连接状态"""
|
||||
# 禁用连接按钮，防止重复点击
|
||||
self.check_button.configure(state="disabled")
|
||||
self.refresh_button.configure(state="disabled")
|
||||
|
||||
# 更新状态栏
|
||||
self.status_var.set("正在检测连接...")
|
||||
self.root.update()
|
||||
|
||||
# 使用线程进行连接检测
|
||||
def check_connections():
|
||||
try:
|
||||
# 检测TV连接
|
||||
ucd_connected = self.check_port_connection(is_ucd=True)
|
||||
self.root.after(
|
||||
0,
|
||||
lambda: self.update_connection_indicator(
|
||||
self.ucd_status_indicator, ucd_connected
|
||||
),
|
||||
)
|
||||
|
||||
# 检测CA连接
|
||||
ca_connected = self.check_port_connection(is_ucd=False)
|
||||
self.root.after(
|
||||
0,
|
||||
lambda: self.update_connection_indicator(
|
||||
self.ca_status_indicator, ca_connected
|
||||
),
|
||||
)
|
||||
|
||||
# 更新状态栏
|
||||
self.root.after(0, lambda: self.status_var.set("连接检测完成"))
|
||||
|
||||
# 重新启用所有控件
|
||||
self.root.after(0, self.enable_com_widgets)
|
||||
except Exception as e:
|
||||
self.root.after(0, lambda: self.log_gui.log(f"连接检测出错: {e}"))
|
||||
self.root.after(0, self.enable_com_widgets)
|
||||
|
||||
# 启动线程
|
||||
threading.Thread(target=check_connections, daemon=True).start()
|
||||
|
||||
return _dev_check_com_connections(self)
|
||||
def update_connection_indicator(self, indicator, connected):
|
||||
"""更新连接状态指示器颜色"""
|
||||
if connected:
|
||||
indicator.config(bg="green")
|
||||
else:
|
||||
indicator.config(bg="red")
|
||||
|
||||
return _dev_update_connection_indicator(self, indicator, connected)
|
||||
def check_port_connection(self, is_ucd=True):
|
||||
"""检测指定端口是否可以连接"""
|
||||
try:
|
||||
if is_ucd:
|
||||
if self.ucd.status:
|
||||
try:
|
||||
self.ucd.close()
|
||||
except:
|
||||
pass
|
||||
if not self.ucd.open(self.ucd_list_var.get()):
|
||||
self.log_gui.log(
|
||||
f"设备 {self.ucd_list_var.get()} 异常，UCD323连接失败"
|
||||
)
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
else:
|
||||
# 如果CA对象已存在，先关闭
|
||||
if self.ca is not None:
|
||||
try:
|
||||
self.ca.close()
|
||||
except:
|
||||
pass
|
||||
channel_value = self.ca_channel_var.get()
|
||||
str_channel = f"{int(channel_value):02d}"
|
||||
self.ca = CASerail()
|
||||
self.ca.open(self.config.device_config["ca_com"], 19200, 7, "E", 2)
|
||||
# data = self.ca.set_xyLv_Display()
|
||||
data = self.ca.set_all_Display()
|
||||
if data:
|
||||
data = self.ca.setSynchMode(3)
|
||||
data = self.ca.setMeasureSpeed(1)
|
||||
if True:
|
||||
time.sleep(0.5)
|
||||
data = self.ca.setZeroCalibration()
|
||||
channel_value = self.ca_channel_var.get()
|
||||
str_channel = f"{int(channel_value):02d}"
|
||||
data = self.ca.setChannel(str_channel)
|
||||
return True
|
||||
else:
|
||||
self.log_gui.log(
|
||||
f"端口 {self.config.device_config["ca_com"]} 异常，色温仪连接失败"
|
||||
)
|
||||
self.ca.close()
|
||||
self.ca = None
|
||||
return False
|
||||
except Exception as e:
|
||||
self.log_gui.log(f"端口连接失败: {e}")
|
||||
return False
|
||||
|
||||
return _dev_check_port_connection(self, is_ucd)
|
||||
def enable_com_widgets(self):
|
||||
"""重新启用所有控件"""
|
||||
self.check_button.configure(state="normal")
|
||||
self.refresh_button.configure(state="normal")
|
||||
|
||||
return _dev_enable_com_widgets(self)
|
||||
def disconnect_com_connections(self):
|
||||
"""断开所有串口连接"""
|
||||
try:
|
||||
# 断开TV连接
|
||||
if self.ucd.status:
|
||||
try:
|
||||
self.ucd.close()
|
||||
except:
|
||||
pass
|
||||
finally:
|
||||
self.ucd.status = False
|
||||
self.log_gui.log("UCD连接已断开")
|
||||
|
||||
# 断开CA连接
|
||||
if self.ca is not None:
|
||||
try:
|
||||
self.ca.close()
|
||||
except:
|
||||
pass
|
||||
finally:
|
||||
self.ca = None
|
||||
self.log_gui.log("CA连接已断开")
|
||||
|
||||
# 重新启用相关控件
|
||||
self.enable_com_widgets()
|
||||
self.ucd_status_indicator.config(bg="gray")
|
||||
self.ca_status_indicator.config(bg="gray")
|
||||
self.status_var.set("串口连接已断开")
|
||||
|
||||
except Exception as e:
|
||||
self.log_gui.log(f"断开连接时发生错误: {str(e)}")
|
||||
messagebox.showerror("错误", f"断开连接失败: {str(e)}")
|
||||
|
||||
return _dev_disconnect_com_connections(self)
|
||||
def create_test_type_frame(self):
|
||||
"""创建测试类型选择区域（侧边栏形式）"""
|
||||
# 设置测试类型变量
|
||||
@@ -6517,34 +6356,11 @@ class PQAutomationApp:
|
||||
self.log_gui.log(f"❌ 屏模组信号格式更改失败: {str(e)}")
|
||||
|
||||
def load_pq_config(self):
|
||||
"""加载PQ配置（兼容打包后的程序）"""
|
||||
try:
|
||||
# ✅ 使用 self.config_file（已经是动态路径）
|
||||
if os.path.exists(self.config_file):
|
||||
with open(self.config_file, "r", encoding="utf-8") as f:
|
||||
config_dict = json.load(f)
|
||||
self.config.from_dict(config_dict)
|
||||
if hasattr(self, "log_gui"):
|
||||
self.log_gui.log("✓ 配置文件加载成功")
|
||||
else:
|
||||
if hasattr(self, "log_gui"):
|
||||
self.log_gui.log("⚠️ 配置文件不存在，使用默认配置")
|
||||
except Exception as e:
|
||||
if hasattr(self, "log_gui"):
|
||||
self.log_gui.log(f"⚠️ 加载配置文件失败: {str(e)}，使用默认配置")
|
||||
|
||||
"""转发到 app.config_io.load_pq_config（Step 4 重构）"""
|
||||
return _cfg_load_pq_config(self)
|
||||
def save_pq_config(self):
|
||||
"""保存PQ配置（兼容打包后的程序）"""
|
||||
try:
|
||||
# 确保目录存在
|
||||
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
|
||||
|
||||
# 保存配置
|
||||
self.config.save_to_file(self.config_file)
|
||||
except Exception as e:
|
||||
if hasattr(self, "log_gui"):
|
||||
self.log_gui.log(f"保存配置文件失败: {str(e)}")
|
||||
|
||||
"""转发到 app.config_io.save_pq_config（Step 4 重构）"""
|
||||
return _cfg_save_pq_config(self)
|
||||
def on_closing(self):
|
||||
"""窗口关闭时的处理"""
|
||||
try:
|
||||
@@ -6842,272 +6658,29 @@ class PQAutomationApp:
|
||||
self.log_gui.log("✓ HDR 单步调试面板已打开（独立窗口）")
|
||||
|
||||
def clear_config_file(self):
|
||||
"""清理配置文件（兼容打包后的程序）"""
|
||||
from tkinter import messagebox
|
||||
|
||||
config_file = self.get_config_path()
|
||||
|
||||
try:
|
||||
if os.path.exists(config_file):
|
||||
os.remove(config_file)
|
||||
self.config_cleared = True
|
||||
messagebox.showinfo("提示", "✓ 清理成功")
|
||||
self.log_gui.log("✓ 配置文件清理成功")
|
||||
else:
|
||||
messagebox.showinfo("提示", "配置文件不存在")
|
||||
self.log_gui.log("⚠️ 配置文件不存在")
|
||||
|
||||
except Exception as e:
|
||||
messagebox.showerror("错误", "❌ 清理失败")
|
||||
self.log_gui.log(f"❌ 配置文件清理失败: {str(e)}")
|
||||
|
||||
"""转发到 app.config_io.clear_config_file（Step 4 重构）"""
|
||||
return _cfg_clear_config_file(self)
|
||||
def start_local_dimming_test(self):
|
||||
"""开始 Local Dimming 测试"""
|
||||
# 检查设备连接
|
||||
if not self.ca or not self.ucd.status:
|
||||
messagebox.showerror("错误", "请先连接 CA410 和 UCD323")
|
||||
return
|
||||
|
||||
# 禁用按钮
|
||||
self.ld_start_btn.config(state=tk.DISABLED)
|
||||
self.ld_stop_btn.config(state=tk.NORMAL)
|
||||
self.ld_save_btn.config(state=tk.DISABLED)
|
||||
|
||||
# 清空结果
|
||||
for item in self.ld_tree.get_children():
|
||||
self.ld_tree.delete(item)
|
||||
|
||||
# 获取配置
|
||||
wait_time = float(self.ld_wait_time_var.get())
|
||||
|
||||
# 在新线程中执行测试
|
||||
def run_test():
|
||||
from utils.local_dimming_test import LocalDimmingTest, LocalDimmingController
|
||||
|
||||
# 从设备当前 timing 获取分辨率
|
||||
ld_ctrl = LocalDimmingController(self.ucd)
|
||||
cur_w, cur_h = ld_ctrl.get_current_resolution()
|
||||
resolution = f"{cur_w}x{cur_h}"
|
||||
|
||||
ld_test = LocalDimmingTest(
|
||||
self.ucd,
|
||||
self.ca,
|
||||
log_callback=self.log_gui.log,
|
||||
)
|
||||
|
||||
ld_test.wait_time = wait_time
|
||||
|
||||
results = ld_test.run_test(resolution=resolution)
|
||||
|
||||
# 保存到实例变量
|
||||
self.ld_test_instance = ld_test
|
||||
self.ld_test_results = results
|
||||
|
||||
# 更新结果显示
|
||||
self.root.after(0, lambda: self.update_ld_results(results))
|
||||
|
||||
# 清理临时文件
|
||||
ld_test.cleanup()
|
||||
|
||||
# 恢复按钮状态
|
||||
self.root.after(0, lambda: self.ld_start_btn.config(state=tk.NORMAL))
|
||||
self.root.after(0, lambda: self.ld_stop_btn.config(state=tk.DISABLED))
|
||||
self.root.after(0, lambda: self.ld_save_btn.config(state=tk.NORMAL))
|
||||
|
||||
threading.Thread(target=run_test, daemon=True).start()
|
||||
|
||||
"""转发到 app.tests.local_dimming.start_local_dimming_test（Step 4 重构）"""
|
||||
return _ld_start_local_dimming_test(self)
|
||||
def update_ld_results(self, results):
|
||||
"""更新 Local Dimming 结果显示"""
|
||||
for percentage, x, y, lv, X, Y, Z in results:
|
||||
self.ld_tree.insert(
|
||||
"",
|
||||
tk.END,
|
||||
values=(f"{percentage}%", f"{lv:.2f}", f"{x:.4f}", f"{y:.4f}"),
|
||||
)
|
||||
|
||||
"""转发到 app.tests.local_dimming.update_ld_results（Step 4 重构）"""
|
||||
return _ld_update_ld_results(self, results)
|
||||
def stop_local_dimming_test(self):
|
||||
"""停止 Local Dimming 测试"""
|
||||
if hasattr(self, "ld_test_instance"):
|
||||
self.ld_test_instance.stop()
|
||||
|
||||
def save_local_dimming_results(self):
|
||||
"""保存 Local Dimming 结果"""
|
||||
from tkinter import filedialog
|
||||
import csv
|
||||
import datetime
|
||||
|
||||
default_name = f"LocalDimming_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
|
||||
|
||||
save_path = filedialog.asksaveasfilename(
|
||||
title="保存测试结果",
|
||||
initialfile=default_name,
|
||||
defaultextension=".csv",
|
||||
filetypes=[("CSV 文件", "*.csv"), ("所有文件", "*.*")],
|
||||
)
|
||||
|
||||
if not save_path:
|
||||
return
|
||||
|
||||
try:
|
||||
with open(save_path, "w", newline="", encoding="utf-8-sig") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["窗口百分比", "x", "y", "亮度 (cd/m²)", "X", "Y", "Z"])
|
||||
|
||||
for item in self.ld_tree.get_children():
|
||||
values = self.ld_tree.item(item, "values")
|
||||
# 从 self.ld_test_results 获取完整数据
|
||||
percentage_str = values[0]
|
||||
percentage = int(percentage_str.replace("%", ""))
|
||||
|
||||
# 找到对应的完整数据
|
||||
for p, x, y, lv, X, Y, Z in self.ld_test_results:
|
||||
if p == percentage:
|
||||
writer.writerow([f"{p}%", x, y, lv, X, Y, Z])
|
||||
break
|
||||
|
||||
self.log_gui.log(f"✓ 测试结果已保存: {save_path}")
|
||||
messagebox.showinfo("成功", f"测试结果已保存到:\n{save_path}")
|
||||
|
||||
except Exception as e:
|
||||
self.log_gui.log(f"❌ 保存失败: {str(e)}")
|
||||
messagebox.showerror("错误", f"保存失败: {str(e)}")
|
||||
|
||||
"""转发到 app.tests.local_dimming.stop_local_dimming_test（Step 4 重构）"""
|
||||
return _ld_stop_local_dimming_test(self)
|
||||
def send_ld_window(self, percentage):
|
||||
"""发送指定百分比的窗口"""
|
||||
if not self.ucd.status:
|
||||
messagebox.showwarning("警告", "请先连接 UCD323 设备")
|
||||
return
|
||||
|
||||
self.log_gui.log(f"🔆 发送 {percentage}% 窗口...")
|
||||
|
||||
# 记录当前百分比（用于测量）
|
||||
self.current_ld_percentage = percentage
|
||||
|
||||
def send():
|
||||
from utils.local_dimming_test import LocalDimmingController
|
||||
|
||||
ld_controller = LocalDimmingController(self.ucd)
|
||||
|
||||
# 从设备当前 timing 获取分辨率
|
||||
width, height = ld_controller.get_current_resolution()
|
||||
|
||||
# 生成并发送图片
|
||||
success = ld_controller.send_window_pattern_with_resolution(
|
||||
percentage, width, height
|
||||
)
|
||||
|
||||
if success:
|
||||
self.root.after(
|
||||
0, lambda: self.log_gui.log(f"✅ {percentage}% 窗口已发送")
|
||||
)
|
||||
else:
|
||||
self.root.after(
|
||||
0, lambda: self.log_gui.log(f"❌ {percentage}% 窗口发送失败")
|
||||
)
|
||||
|
||||
threading.Thread(target=send, daemon=True).start()
|
||||
|
||||
"""转发到 app.tests.local_dimming.send_ld_window（Step 4 重构）"""
|
||||
return _ld_send_ld_window(self, percentage)
|
||||
def measure_ld_luminance(self):
|
||||
"""测量当前亮度"""
|
||||
if not self.ca:
|
||||
messagebox.showwarning("警告", "请先连接 CA410 色度计")
|
||||
return
|
||||
|
||||
if self.current_ld_percentage is None:
|
||||
messagebox.showinfo("提示", "请先发送一个窗口图案")
|
||||
return
|
||||
|
||||
self.log_gui.log("📏 正在采集亮度...")
|
||||
|
||||
def measure():
|
||||
try:
|
||||
x, y, lv, X, Y, Z = self.ca.readAllDisplay()
|
||||
|
||||
if lv is not None:
|
||||
import datetime
|
||||
|
||||
timestamp = datetime.datetime.now().strftime("%H:%M:%S")
|
||||
|
||||
# 更新显示
|
||||
self.root.after(
|
||||
0,
|
||||
lambda: self.ld_result_label.config(
|
||||
text=f"亮度: {lv:.2f} cd/m² | x: {x:.4f} | y: {y:.4f}"
|
||||
),
|
||||
)
|
||||
|
||||
# 添加到表格
|
||||
self.root.after(
|
||||
0,
|
||||
lambda: self.ld_tree.insert(
|
||||
"",
|
||||
tk.END,
|
||||
values=(
|
||||
f"{self.current_ld_percentage}%",
|
||||
f"{lv:.2f}",
|
||||
f"{x:.4f}",
|
||||
f"{y:.4f}",
|
||||
timestamp,
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
self.root.after(
|
||||
0, lambda: self.log_gui.log(f"✅ 采集完成: {lv:.2f} cd/m²")
|
||||
)
|
||||
else:
|
||||
self.root.after(0, lambda: self.log_gui.log("❌ 采集失败"))
|
||||
|
||||
except Exception as e:
|
||||
self.root.after(0, lambda: self.log_gui.log(f"❌ 采集异常: {str(e)}"))
|
||||
|
||||
threading.Thread(target=measure, daemon=True).start()
|
||||
|
||||
"""转发到 app.tests.local_dimming.measure_ld_luminance（Step 4 重构）"""
|
||||
return _ld_measure_ld_luminance(self)
|
||||
def clear_ld_records(self):
|
||||
"""清空测试记录"""
|
||||
for item in self.ld_tree.get_children():
|
||||
self.ld_tree.delete(item)
|
||||
self.ld_result_label.config(text="亮度: -- cd/m² | x: -- | y: --")
|
||||
self.current_ld_percentage = None
|
||||
self.log_gui.log("🗑️ 测试记录已清空")
|
||||
|
||||
"""转发到 app.tests.local_dimming.clear_ld_records（Step 4 重构）"""
|
||||
return _ld_clear_ld_records(self)
|
||||
def save_local_dimming_results(self):
|
||||
"""保存 Local Dimming 结果"""
|
||||
from tkinter import filedialog
|
||||
import csv
|
||||
import datetime
|
||||
|
||||
if len(self.ld_tree.get_children()) == 0:
|
||||
messagebox.showinfo("提示", "没有可保存的数据")
|
||||
return
|
||||
|
||||
default_name = f"LocalDimming_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
|
||||
|
||||
save_path = filedialog.asksaveasfilename(
|
||||
title="保存测试结果",
|
||||
initialfile=default_name,
|
||||
defaultextension=".csv",
|
||||
filetypes=[("CSV 文件", "*.csv"), ("所有文件", "*.*")],
|
||||
)
|
||||
|
||||
if not save_path:
|
||||
return
|
||||
|
||||
try:
|
||||
with open(save_path, "w", newline="", encoding="utf-8-sig") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["窗口百分比", "亮度 (cd/m²)", "x", "y", "时间"])
|
||||
|
||||
for item in self.ld_tree.get_children():
|
||||
values = self.ld_tree.item(item, "values")
|
||||
writer.writerow(values)
|
||||
|
||||
self.log_gui.log(f"✓ 测试结果已保存: {save_path}")
|
||||
messagebox.showinfo("成功", f"测试结果已保存到:\n{save_path}")
|
||||
|
||||
except Exception as e:
|
||||
self.log_gui.log(f"❌ 保存失败: {str(e)}")
|
||||
messagebox.showerror("错误", f"保存失败: {str(e)}")
|
||||
"""转发到 app.tests.local_dimming.save_local_dimming_results（Step 4 重构）"""
|
||||
return _ld_save_local_dimming_results(self)
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
Reference in New Issue
Block a user