MOKA/pqAutomationApp
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改calman灰阶点击异常、修改色准结果显示异常

Browse Source
This commit is contained in:
xinzhu.yin
2026-06-02 17:34:46 +08:00
parent 85ac47e8de
commit 3aa975c4d3
19 changed files with 968 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
drivers/UCD323_Enum.py
View File

@@ -110,7 +110,16 @@ class UCDEnum:
}
if not colorimetry_str:
return None
return colorimetry_map.get(colorimetry_str.lower(), None)
# Normalize: strip hyphens, spaces, dots, underscores so that
# "DCI-P3" → "dcip3", "BT.709" → "bt709", "BT.2020 YCbCr" → "bt2020ycbcr"
normalized = (
colorimetry_str.lower()
.replace("-", "")
.replace(" ", "")
.replace(".", "")
.replace("_", "")
)
return colorimetry_map.get(normalized, colorimetry_map.get(colorimetry_str.lower(), None))
class VideoPatternInfo:
class VideoPattern(IntEnum):

202
drivers/UCD323_Function.py
View File

@@ -1,10 +1,14 @@
# -*- coding: UTF-8 -*-
import logging
import UniTAP
import time
import gc
from drivers.UCD323_Enum import UCDEnum
log = logging.getLogger(__name__)
class UCDController:
"""UCD323信号发生器控制类"""
@@ -23,6 +27,7 @@ class UCDController:
self.current_pattern_param = None
self.current_pattern_params = None
self.current_pattern_index = 0
self.last_error = None
def search_device(self):
"""搜索可用设备"""
@@ -140,6 +145,7 @@ class UCDController:
raise RuntimeError("UCD 未打开，无法获取 TX 模块")
interface = getattr(self, "current_interface", None)
log.info("UCDController.get_tx_modules interface=%s", interface)
if interface in (None, "HDMI"):
return self.role.hdtx.pg, self.role.hdtx.ag
if interface in ("DP", "Type-C"):
@@ -189,6 +195,7 @@ class UCDController:
def set_ucd_params(self, config):
"""设置UCD323参数"""
self.last_error = None
self.config = config
test_type = self.config.current_test_type
@@ -200,16 +207,34 @@ class UCDController:
colorimetry = self.config.current_test_types[test_type]["colorimetry"]
if not self.set_color_mode(color_format, bpc, colorimetry):
self.last_error = (
f"set_color_mode failed: color_format={color_format}, bpc={bpc}, colorimetry={colorimetry}"
)
log.error(
"UCDController.set_ucd_params set_color_mode failed test_type=%s color_format=%s bpc=%s colorimetry=%s",
test_type,
color_format,
bpc,
colorimetry,
)
return False
timing_str = self.config.current_test_types[test_type]["timing"]
self.set_timing_from_string(timing_str)
if not self.set_timing_from_string(timing_str):
self.last_error = f"set_timing_from_string failed: timing={timing_str}"
log.error(
"UCDController.set_ucd_params set_timing_from_string failed test_type=%s timing=%s",
test_type,
timing_str,
)
return False
self.current_pattern_index = 0
pattern_mode = self.config.current_pattern["pattern_mode"]
pattern = UCDEnum.VideoPatternInfo.get_video_pattern(pattern_mode)
if pattern is None:
self.last_error = f"get_video_pattern failed: pattern_mode={pattern_mode}"
return False
self.current_pattern = pattern
@@ -219,10 +244,17 @@ class UCDController:
def run(self):
"""运行设备"""
log.info(
"UCDController.run start current_pattern=%s has_pattern_param=%s",
getattr(self.current_pattern, "name", self.current_pattern),
self.current_pattern_param is not None,
)
self.apply_video_mode()
self.apply_pattern()
pg, _ = self.get_tx_modules()
log.info("UCDController.run calling pg.apply()")
pg.apply()
log.info("UCDController.run done")
return True
def send_image_pattern(self, image_path):
@@ -245,12 +277,15 @@ class UCDController:
return False
try:
log.info("UCDController.send_solid_rgb_pattern rgb=%s", rgb)
self.current_pattern = UCDEnum.VideoPatternInfo.get_video_pattern("solidcolor")
if self.current_pattern is None:
log.error("UCDController.send_solid_rgb_pattern failed: solidcolor pattern not found")
return False
return self.send_current_pattern_params(list(rgb))
except Exception:
log.exception("UCDController.send_solid_rgb_pattern exception")
return False
def send_current_pattern_params(self, pattern_params):
@@ -260,17 +295,27 @@ class UCDController:
try:
if self.current_pattern is None:
log.error("UCDController.send_current_pattern_params failed: current_pattern is None")
return False
log.info(
"UCDController.send_current_pattern_params pattern=%s params=%s",
getattr(self.current_pattern, "name", self.current_pattern),
pattern_params,
)
if pattern_params is not None and not self.set_pattern(
self.current_pattern,
pattern_params,
):
log.error("UCDController.send_current_pattern_params failed: set_pattern returned False")
return False
log.info("UCDController.send_current_pattern_params calling run()")
self.run()
log.info("UCDController.send_current_pattern_params done")
return True
except Exception:
log.exception("UCDController.send_current_pattern_params exception")
return False
def set_color_mode(self, cf, bpc, cm):
@@ -298,8 +343,11 @@ class UCDController:
def apply_video_mode(self):
"""应用当前 color_info 和 timing"""
if self.current_timing:
log.info("UCDController.apply_video_mode start timing=%s", self.current_timing)
self.set_video_mode()
log.info("UCDController.apply_video_mode done")
return True
log.warning("UCDController.apply_video_mode skipped: current_timing is None")
return False
def set_video_mode(self):
@@ -313,28 +361,48 @@ class UCDController:
self.color_info.bpc,
)
self.format_changed = (current_config != getattr(self, "_last_sent_config", None))
log.info(
"UCDController.set_video_mode format_changed=%s color_format=%s colorimetry=%s dynamic_range=%s bpc=%s",
self.format_changed,
self.color_info.color_format,
self.color_info.colorimetry,
self.color_info.dynamic_range,
self.color_info.bpc,
)
video_mode = UniTAP.VideoMode(
timing=self.current_timing, color_info=self.color_info
)
pg, _ = self.get_tx_modules()
log.info("UCDController.set_video_mode calling pg.set_vm()")
pg.set_vm(vm=video_mode)
log.info("UCDController.set_video_mode done")
self._last_sent_config = current_config
return True
def set_pattern(self, pattern, pattern_params=None):
"""设置pattern"""
if self.current_timing:
needs_params = {
UCDEnum.VideoPatternInfo.VideoPatternParams.SolidColor,
UCDEnum.VideoPatternInfo.VideoPatternParams.WhiteVStrips,
UCDEnum.VideoPatternInfo.VideoPatternParams.GradientRGBStripes,
UCDEnum.VideoPatternInfo.VideoPatternParams.MotionPattern,
UCDEnum.VideoPatternInfo.VideoPatternParams.SquareWindow,
}
if pattern in needs_params and pattern_params:
self.set_pattern_params(pattern, pattern_params)
return True
return False
if self.current_timing is None:
# Pattern-only updates (e.g. Calman patch click) can still be applied on
# an already active output mode. Missing timing should not block pattern staging.
log.warning("UCDController.set_pattern current_timing is None; continue with pattern-only apply")
needs_params = {
UCDEnum.VideoPatternInfo.VideoPatternParams.SolidColor,
UCDEnum.VideoPatternInfo.VideoPattern.SolidColor,
UCDEnum.VideoPatternInfo.VideoPatternParams.WhiteVStrips,
UCDEnum.VideoPatternInfo.VideoPatternParams.GradientRGBStripes,
UCDEnum.VideoPatternInfo.VideoPatternParams.MotionPattern,
UCDEnum.VideoPatternInfo.VideoPatternParams.SquareWindow,
}
log.info(
"UCDController.set_pattern pattern=%s pattern_params=%s needs_params=%s",
getattr(pattern, "name", pattern),
pattern_params,
pattern in needs_params,
)
if pattern in needs_params and pattern_params is not None:
self.set_pattern_params(pattern, pattern_params)
return True
def set_next_pattern(self):
"""设置下一个pattern"""
@@ -350,25 +418,40 @@ class UCDController:
def set_pattern_params(self, pattern, pattern_params):
"""设置pattern参数"""
if pattern:
if pattern == UCDEnum.VideoPatternInfo.VideoPatternParams.SolidColor:
if pattern is not None:
solid_color_patterns = {
UCDEnum.VideoPatternInfo.VideoPatternParams.SolidColor,
UCDEnum.VideoPatternInfo.VideoPattern.SolidColor,
}
if pattern in solid_color_patterns:
log.info("UCDController.set_pattern_params solid_color rgb=%s", pattern_params)
self.current_pattern_param = UniTAP.SolidColorParams(
first=pattern_params[0],
second=pattern_params[1],
third=pattern_params[2],
)
return True
log.warning("UCDController.set_pattern_params unsupported pattern=%s", getattr(pattern, "name", pattern))
return False
def apply_pattern(self):
"""应用当前pattern"""
if self.current_pattern:
if self.current_pattern is not None:
log.info(
"UCDController.apply_pattern start pattern=%s has_params=%s",
getattr(self.current_pattern, "name", self.current_pattern),
self.current_pattern_param is not None,
)
pg, _ = self.get_tx_modules()
log.info("UCDController.apply_pattern calling pg.set_pattern()")
pg.set_pattern(self.current_pattern)
if self.current_pattern_param:
if self.current_pattern_param is not None:
log.info("UCDController.apply_pattern calling pg.set_pattern_params()")
pg.set_pattern_params(self.current_pattern_param)
log.info("UCDController.apply_pattern done")
return True
log.warning("UCDController.apply_pattern skipped: current_pattern is None")
return False
def search_timing(self, width, height, refresh_rate, resolution_type=None):
@@ -383,15 +466,30 @@ class UCDController:
elif resolution_type == "cvt":
standard = UniTAP.common.timing.Timing.Standard.SD_CVT
timing = self.timing_manager.search(
h_active=width,
v_active=height,
f_rate=int(refresh_rate) * 1000,
standard=standard,
)
rr = float(refresh_rate)
# Try both exact and NTSC-compatible rates (e.g. 120000 / 119880).
f_rate_candidates = [
int(round(rr * 1000)),
int(rr * 1000),
int(round((rr * 1000.0) * 1000.0 / 1001.0)),
]
# 去重并保持顺序
f_rate_candidates = list(dict.fromkeys(f_rate_candidates))
if timing:
return timing
standards = [standard]
if standard is not None:
standards.append(None)
for std in standards:
for f_rate in f_rate_candidates:
timing = self.timing_manager.search(
h_active=width,
v_active=height,
f_rate=f_rate,
standard=std,
)
if timing:
return timing
else:
for res_type in ["dmt", "cta", "cvt", "ovt"]:
result = self.search_timing(width, height, refresh_rate, res_type)
@@ -492,18 +590,54 @@ class UCDController:
def set_timing_from_string(self, timing_str):
"""根据格式化timing字符串设置设备timing"""
spec = self.parse_formatted_timing(timing_str)
try:
spec = self.parse_formatted_timing(timing_str)
except Exception:
log.exception("UCDController.set_timing_from_string parse failed timing=%s", timing_str)
return False
rtype = spec["resolution_type"]
rid = spec.get("resolution_id")
width = spec["width"]
height = spec["height"]
fr = spec["refresh_rate"]
if rtype != "ovt":
timing = self.search_timing(width, height, fr, rtype)
if timing:
self.current_timing = timing
return True
if rid is not None and self.set_timing_from_id(rtype, rid):
log.info(
"UCDController.set_timing_from_string success by id timing=%s parsed=(%s id=%s)",
timing_str,
rtype,
rid,
)
return True
# Respect selected timing family first (DMT/CTA/CVT/OVT).
timing = self.search_timing(width, height, fr, rtype)
if timing is None:
# Fallback only for robustness: some SDKs may not classify a timing
# exactly as requested family even though width/height/fps matches.
timing = self.search_timing(width, height, fr, None)
if timing:
self.current_timing = timing
log.info(
"UCDController.set_timing_from_string success timing=%s parsed=(%s %sx%s@%s)",
timing_str,
rtype,
width,
height,
fr,
)
return True
log.error(
"UCDController.set_timing_from_string no timing matched timing=%s parsed=(%s %sx%s@%s)",
timing_str,
rtype,
width,
height,
fr,
)
return False
def set_timing_from_id(self, rtype, rid):
@@ -515,6 +649,12 @@ class UCDController:
timing = self.timing_manager.get_cta(rid)
elif rtype.lower() == "cvt":
timing = self.timing_manager.get_cvt(rid)
elif rtype.lower() == "ovt":
get_ovt = getattr(self.timing_manager, "get_ovt", None)
if callable(get_ovt):
timing = get_ovt(rid)
else:
return False
else:
raise ValueError(f"不支持的分辨率类型: {rtype}")

71
drivers/ucd_driver.py
View File

@@ -18,6 +18,7 @@ SDK 调用直接搬入本模块，届时可删除旧 ``UCDController`` 文件。
from __future__ import annotations
from contextlib import contextmanager
import logging
import threading
from abc import ABC, abstractmethod
@@ -48,6 +49,8 @@ if TYPE_CHECKING:
log = logging.getLogger(__name__)
_DEVICE_LOCK_TIMEOUT_SECONDS = 8.0
# ─── §1 DeviceInfo / list_devices ────────────────────────────────
@@ -154,12 +157,50 @@ class UCD323Device(IUcdDevice):
self._info: DeviceInfo | None = None
self._interface: Interface = Interface.HDMI
self._lock = threading.RLock()
self._lock_owner_tid: int | None = None
self._lock_owner_name: str | None = None
self._curr_signal: SignalFormat | None = None
self._curr_timing: TimingSpec | None = None
self._curr_pattern: PatternSpec | None = None
self._last_applied: tuple[SignalFormat, TimingSpec] | None = None
@contextmanager
def _acquire_device_lock(self, op_name: str):
current = threading.current_thread()
log.info(
"UCD323Device.%s acquiring lock timeout=%.1fs tid=%s thread=%s owner_tid=%s owner_thread=%s",
op_name,
_DEVICE_LOCK_TIMEOUT_SECONDS,
threading.get_ident(),
current.name,
self._lock_owner_tid,
self._lock_owner_name,
)
acquired = self._lock.acquire(timeout=_DEVICE_LOCK_TIMEOUT_SECONDS)
if not acquired:
raise UcdStateError(
"UCD device busy: lock timeout in "
f"UCD323Device.{op_name}, "
f"owner_tid={self._lock_owner_tid}, owner_thread={self._lock_owner_name}"
)
prev_owner_tid = self._lock_owner_tid
prev_owner_name = self._lock_owner_name
self._lock_owner_tid = threading.get_ident()
self._lock_owner_name = current.name
log.info(
"UCD323Device.%s lock acquired tid=%s thread=%s",
op_name,
self._lock_owner_tid,
self._lock_owner_name,
)
try:
yield
finally:
self._lock_owner_tid = prev_owner_tid
self._lock_owner_name = prev_owner_name
self._lock.release()
# -- 读访问 --------------------------------------------------
@property
@@ -181,7 +222,7 @@ class UCD323Device(IUcdDevice):
# -- 生命周期 ------------------------------------------------
def open(self, info: DeviceInfo, *, interface: Interface = Interface.HDMI) -> None:
with self._lock:
with self._acquire_device_lock("open"):
assert_transition(self._state, UcdState.OPENED)
if interface is not Interface.HDMI:
# Phase 1：底层 UCDController.open() 写死了 HDMISource。
@@ -200,7 +241,7 @@ class UCD323Device(IUcdDevice):
self._bus.publish(ConnectionChanged(True, info.serial))
def close(self) -> None:
with self._lock:
with self._acquire_device_lock("close"):
if self._state == UcdState.CLOSED:
return
try:
@@ -219,7 +260,7 @@ class UCD323Device(IUcdDevice):
# -- 配置 ----------------------------------------------------
def configure(self, signal: SignalFormat, timing: TimingSpec) -> bool:
with self._lock:
with self._acquire_device_lock("configure"):
if self._state == UcdState.CLOSED:
raise UcdNotConnected("UCD 未连接，无法 configure")
try:
@@ -249,7 +290,7 @@ class UCD323Device(IUcdDevice):
return (signal, timing) != self._last_applied
def set_pattern(self, pattern: PatternSpec) -> None:
with self._lock:
with self._acquire_device_lock("set_pattern"):
# Phase 2 过渡：允许从 OPENED 直接 set_pattern——遗留路径
# （test_runner 等）通过旧 controller.apply_signal_format 写入
# 信号格式，未经过本设备的 configure。此时 self._state 仍为
@@ -260,7 +301,7 @@ class UCD323Device(IUcdDevice):
# 仅本地暂存，真正写硬件在 apply()
def apply(self) -> None:
with self._lock:
with self._acquire_device_lock("apply"):
if self._state == UcdState.CLOSED:
raise UcdNotConnected("UCD 未连接，无法 apply")
if self._curr_pattern is None:
@@ -268,7 +309,9 @@ class UCD323Device(IUcdDevice):
try:
ok = self._apply_pattern_via_controller(self._curr_pattern)
except Exception as exc: # noqa: BLE001
raise UcdSdkError("apply 异常") from exc
raise UcdSdkError(
f"apply 异常: {type(exc).__name__}: {exc}"
) from exc
if not ok:
raise UcdApplyFailed(
f"apply 失败: pattern={self._curr_pattern.kind.value}"
@@ -333,7 +376,21 @@ class UCD323Device(IUcdDevice):
if not self._controller.set_pattern(video_pattern, params):
raise UcdApplyFailed("controller.set_pattern 返回 False")
return bool(self._controller.run())
# Skip apply_video_mode() (i.e. pg.set_vm) – the video format is already
# configured by the main signal panel and re-applying it blocks until the
# device re-locks, causing an apparent UI freeze for pattern-only sends.
if not self._controller.apply_pattern():
raise UcdApplyFailed("controller.apply_pattern 返回 False")
if getattr(self._controller, "current_timing", None) is None:
raise UcdConfigError(
"current_timing is None; please apply selected test profile/timing before sending pattern"
)
try:
pg, _ = self._controller.get_tx_modules()
pg.apply()
except Exception as exc:
raise UcdSdkError("pg.apply() 失败") from exc
return True
def _colorimetry_to_legacy_key(signal: SignalFormat) -> str: