根据新版本调整set_pattern：0~65535
This commit is contained in:
xinzhu.yin
@@ -13,6 +13,20 @@ from app.ucd.device import (
|
||||
list_devices,
|
||||
)
|
||||
from app.ucd.service import PatternService, PatternSession, SignalService
|
||||
from app.solid_color_scale import (
|
||||
SOLID_COLOR_MAX,
|
||||
SOLID_COLOR_MIN,
|
||||
SOLID_COLOR_MIN_BPC,
|
||||
byte_to_solid_color,
|
||||
clamp_solid_color,
|
||||
ensure_solid_color_pattern,
|
||||
migrate_pattern_params,
|
||||
pack_solid_color_hw_registers,
|
||||
pct_to_solid_color,
|
||||
solid_color_rgb,
|
||||
solid_color_to_byte,
|
||||
solid_color_to_pct,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"SignalService",
|
||||
@@ -25,4 +39,15 @@ __all__ = [
|
||||
"EventBus",
|
||||
"ConnectionChanged",
|
||||
"DeviceKind",
|
||||
"SOLID_COLOR_MIN",
|
||||
"SOLID_COLOR_MAX",
|
||||
"SOLID_COLOR_MIN_BPC",
|
||||
"clamp_solid_color",
|
||||
"pct_to_solid_color",
|
||||
"solid_color_to_pct",
|
||||
"solid_color_to_byte",
|
||||
"byte_to_solid_color",
|
||||
"solid_color_rgb",
|
||||
"migrate_pattern_params",
|
||||
"ensure_solid_color_pattern",
|
||||
]
|
||||
|
||||
@@ -43,6 +43,8 @@ from app.ucd.domain import (
|
||||
is_ycbcr,
|
||||
)
|
||||
from app.ucd.enum import UCDEnum
|
||||
from app.ucd.pg_status import read_pattern_generator_status_lines
|
||||
from app.solid_color_scale import SOLID_COLOR_MIN_BPC, clamp_solid_color, pack_solid_color_hw_registers
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
_DEVICE_LOCK_TIMEOUT_SECONDS = 8.0
|
||||
@@ -108,6 +110,7 @@ class _UcdSdkBackend:
|
||||
self.current_pattern_params = None
|
||||
self.current_pattern_index = 0
|
||||
self.last_error = None
|
||||
self._solid_color_stream_ready = False
|
||||
|
||||
def search_device(self):
|
||||
"""搜索可用设备"""
|
||||
@@ -213,6 +216,10 @@ class _UcdSdkBackend:
|
||||
self.current_pattern_params = None
|
||||
self.current_pattern_index = 0
|
||||
self.current_interface = "HDMI"
|
||||
self._solid_color_stream_ready = False
|
||||
|
||||
def _invalidate_solid_color_stream(self) -> None:
|
||||
self._solid_color_stream_ready = False
|
||||
|
||||
def _force_cleanup(self):
|
||||
"""强制清理所有状态"""
|
||||
@@ -440,6 +447,7 @@ class _UcdSdkBackend:
|
||||
pg, _ = self.get_tx_modules()
|
||||
log.info("UcdSdk.set_video_mode calling pg.set_vm()")
|
||||
pg.set_vm(vm=video_mode)
|
||||
self._invalidate_solid_color_stream()
|
||||
self._stop_audio_output()
|
||||
log.info("UcdSdk.set_video_mode done")
|
||||
self._last_sent_config = current_config
|
||||
@@ -681,6 +689,104 @@ class _UcdSdkBackend:
|
||||
log.warning("UcdSdk.set_pattern_params unsupported pattern=%s", getattr(pattern, "name", pattern))
|
||||
return False
|
||||
|
||||
def _solid_color_pattern_ids(self):
|
||||
return {
|
||||
UCDEnum.VideoPatternInfo.VideoPatternParams.SolidColor,
|
||||
UCDEnum.VideoPatternInfo.VideoPattern.SolidColor,
|
||||
}
|
||||
|
||||
def _is_solid_color_pattern(self, pattern) -> bool:
|
||||
return pattern in self._solid_color_pattern_ids()
|
||||
|
||||
def _write_pg_solid_color_registers(self, pg, rgb_list) -> None:
|
||||
"""直接写入 SolidColor 硬件寄存器（65535 刻度，无 SDK 左移）。"""
|
||||
from ctypes import c_uint
|
||||
|
||||
from UniTAP.libs.lib_tsi import ci
|
||||
|
||||
param0, param1 = pack_solid_color_hw_registers(rgb_list)
|
||||
pg._become_active()
|
||||
pg._PatternGenerator__io.set(ci.TSI_PG_PREDEF_PATTERN_PARAM_1, param0, c_uint)
|
||||
pg._PatternGenerator__io.set(ci.TSI_PG_PREDEF_PATTERN_PARAM_2, param1, c_uint)
|
||||
log.info(
|
||||
"UcdSdk._write_pg_solid_color_registers rgb=%s hw=(0x%08X, 0x%08X)",
|
||||
rgb_list,
|
||||
param0,
|
||||
param1,
|
||||
)
|
||||
|
||||
def apply_solid_color(self, rgb, *, force_reapply: bool = False) -> bool:
|
||||
"""发送 SolidColor，顺序对齐 ``work_with_solid_color.py``。
|
||||
|
||||
首次（或信号格式变化后）：``set_pattern → set_vm → apply() → set_pattern_params``。
|
||||
同一会话内后续仅 ``set_pattern_params``，避免重复 ``apply()`` 导致 HDMI 重锁、
|
||||
灰阶测量出现锯齿形 Gamma 曲线。
|
||||
"""
|
||||
if not self.status or not self.role:
|
||||
return False
|
||||
|
||||
try:
|
||||
pg, _ = self.get_tx_modules()
|
||||
pattern = UCDEnum.VideoPatternInfo.get_video_pattern("solidcolor")
|
||||
if pattern is None:
|
||||
log.error("UcdSdk.apply_solid_color failed: solidcolor pattern not found")
|
||||
return False
|
||||
|
||||
rgb_list = [clamp_solid_color(v) for v in list(rgb)[:3]]
|
||||
if len(rgb_list) < 3:
|
||||
log.error("UcdSdk.apply_solid_color invalid rgb=%s", rgb)
|
||||
return False
|
||||
|
||||
if self.current_timing is None:
|
||||
self.current_timing = self._resolve_timing(pg)
|
||||
if self.current_timing is None:
|
||||
log.error("UcdSdk.apply_solid_color current_timing is None")
|
||||
self.last_error = "current_timing is None"
|
||||
return False
|
||||
|
||||
bpc_bumped = False
|
||||
if self.color_info.bpc < SOLID_COLOR_MIN_BPC:
|
||||
log.info(
|
||||
"UcdSdk.apply_solid_color bump bpc %s -> %s for SolidColor video mode",
|
||||
self.color_info.bpc,
|
||||
SOLID_COLOR_MIN_BPC,
|
||||
)
|
||||
self.color_info.bpc = SOLID_COLOR_MIN_BPC
|
||||
bpc_bumped = True
|
||||
|
||||
self.current_pattern = pattern
|
||||
self.current_pattern_param = UniTAP.SolidColorParams(
|
||||
first=rgb_list[0],
|
||||
second=rgb_list[1],
|
||||
third=rgb_list[2],
|
||||
)
|
||||
|
||||
need_full_setup = (
|
||||
force_reapply
|
||||
or bpc_bumped
|
||||
or not self._solid_color_stream_ready
|
||||
)
|
||||
|
||||
if need_full_setup:
|
||||
log.info("UcdSdk.apply_solid_color full setup rgb=%s", rgb_list)
|
||||
pg.set_pattern(pattern=pattern)
|
||||
self.set_video_mode()
|
||||
if not self._apply_pg_output(pg):
|
||||
log.error("UcdSdk.apply_solid_color pg.apply failed")
|
||||
self._invalidate_solid_color_stream()
|
||||
return False
|
||||
self._solid_color_stream_ready = True
|
||||
else:
|
||||
log.info("UcdSdk.apply_solid_color params-only rgb=%s", rgb_list)
|
||||
|
||||
self._write_pg_solid_color_registers(pg, rgb_list)
|
||||
log.info("UcdSdk.apply_solid_color done")
|
||||
return True
|
||||
except Exception:
|
||||
log.exception("UcdSdk.apply_solid_color exception")
|
||||
self._invalidate_solid_color_stream()
|
||||
return False
|
||||
|
||||
def apply_pattern(self):
|
||||
"""应用当前pattern"""
|
||||
if self.current_pattern is not None:
|
||||
@@ -715,6 +821,7 @@ class _UcdSdkBackend:
|
||||
|
||||
def set_ucd_params(self, config):
|
||||
"""设置UCD323参数"""
|
||||
self._invalidate_solid_color_stream()
|
||||
self.last_error = None
|
||||
self.config = config
|
||||
test_type = self.config.current_test_type
|
||||
@@ -799,6 +906,10 @@ class _UcdSdkBackend:
|
||||
log.error("UcdSdk.send_current_pattern_params failed: set_pattern returned False")
|
||||
return False
|
||||
|
||||
if self._is_solid_color_pattern(self.current_pattern):
|
||||
log.info("UcdSdk.send_current_pattern_params using apply_solid_color")
|
||||
return self.apply_solid_color(pattern_params)
|
||||
|
||||
log.info("UcdSdk.send_current_pattern_params calling run()")
|
||||
self.run()
|
||||
log.info("UcdSdk.send_current_pattern_params done")
|
||||
@@ -814,12 +925,7 @@ class _UcdSdkBackend:
|
||||
|
||||
try:
|
||||
log.info("UcdSdk.send_solid_rgb_pattern rgb=%s", rgb)
|
||||
self.current_pattern = UCDEnum.VideoPatternInfo.get_video_pattern("solidcolor")
|
||||
if self.current_pattern is None:
|
||||
log.error("UcdSdk.send_solid_rgb_pattern failed: solidcolor pattern not found")
|
||||
return False
|
||||
|
||||
return self.send_current_pattern_params(list(rgb))
|
||||
return self.apply_solid_color(list(rgb))
|
||||
except Exception:
|
||||
log.exception("UcdSdk.send_solid_rgb_pattern exception")
|
||||
return False
|
||||
@@ -839,6 +945,20 @@ class _UcdSdkBackend:
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def get_pattern_generator_status_lines(
|
||||
self,
|
||||
pattern_hint: str | None = None,
|
||||
) -> list[str]:
|
||||
"""读取 PG 当前状态，格式对齐 UCD Console 的 Pattern Generator Status。"""
|
||||
if not self.status or not self.role:
|
||||
return []
|
||||
try:
|
||||
pg, _ = self.get_tx_modules()
|
||||
return read_pattern_generator_status_lines(pg, pattern_hint=pattern_hint)
|
||||
except Exception:
|
||||
log.exception("UcdSdk.get_pattern_generator_status_lines failed")
|
||||
return []
|
||||
|
||||
# ─── §3 DeviceInfo / list_devices ────────────────────────────────
|
||||
|
||||
|
||||
@@ -1244,6 +1364,13 @@ class UCD323Device(IUcdDevice):
|
||||
self._state = UcdState.APPLIED
|
||||
return ok
|
||||
|
||||
def get_pattern_generator_status_lines(
|
||||
self,
|
||||
pattern_hint: str | None = None,
|
||||
) -> list[str]:
|
||||
with self._acquire_device_lock("get_pattern_generator_status_lines"):
|
||||
return self._sdk.get_pattern_generator_status_lines(pattern_hint=pattern_hint)
|
||||
|
||||
# --- 内部 ---
|
||||
|
||||
def _apply_dynamic_range(self, signal: SignalFormat) -> None:
|
||||
@@ -1262,6 +1389,13 @@ class UCD323Device(IUcdDevice):
|
||||
raise UcdConfigError("IMAGE pattern 必须提供 image_path")
|
||||
return bool(self._sdk.send_image_pattern(pattern.image_path))
|
||||
|
||||
if pattern.kind is PatternKind.SOLID:
|
||||
if pattern.solid_rgb is None:
|
||||
raise UcdConfigError("SOLID pattern 必须提供 solid_rgb")
|
||||
if not self._sdk.apply_solid_color(list(pattern.solid_rgb)):
|
||||
raise UcdApplyFailed("apply_solid_color 返回 False")
|
||||
return True
|
||||
|
||||
# 预定义图案路径：复用 controller.set_pattern + run()
|
||||
|
||||
video_pattern = UCDEnum.VideoPatternInfo.get_video_pattern(pattern.kind.value)
|
||||
@@ -1270,11 +1404,7 @@ class UCD323Device(IUcdDevice):
|
||||
self._sdk.current_pattern = video_pattern
|
||||
|
||||
params: list[int] | None = None
|
||||
if pattern.kind is PatternKind.SOLID:
|
||||
if pattern.solid_rgb is None:
|
||||
raise UcdConfigError("SOLID pattern 必须提供 solid_rgb")
|
||||
params = list(pattern.solid_rgb)
|
||||
elif pattern.extras:
|
||||
if pattern.extras:
|
||||
params = list(pattern.extras)
|
||||
|
||||
if not self._sdk.set_pattern(video_pattern, params):
|
||||
|
||||
111
app/ucd/pg_status.py
Normal file
111
app/ucd/pg_status.py
Normal file
@@ -0,0 +1,111 @@
|
||||
"""从 UniTAP Pattern Generator 读取状态并格式化为日志行。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
import UniTAP
|
||||
from UniTAP.common import ColorInfo
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _enum_display_name(name: str) -> str:
|
||||
spaced = re.sub(r"([a-z])([A-Z])", r"\1 \2", name)
|
||||
return spaced.replace("_", " ")
|
||||
|
||||
|
||||
def _format_pattern_name(pg, pattern_hint: str | None = None) -> str:
|
||||
if pattern_hint:
|
||||
return pattern_hint
|
||||
|
||||
pattern = getattr(pg, "_pattern", None)
|
||||
if isinstance(pattern, UniTAP.VideoPattern):
|
||||
return _enum_display_name(pattern.name)
|
||||
if isinstance(pattern, str):
|
||||
return os.path.basename(pattern) or "Image File"
|
||||
|
||||
pattern_id = getattr(pg, "_pattern_id", None)
|
||||
if pattern_id is not None:
|
||||
return _enum_display_name(getattr(pattern_id, "name", str(pattern_id)))
|
||||
return "—"
|
||||
|
||||
|
||||
def _format_cef(ci: ColorInfo) -> str:
|
||||
fmt_map = {
|
||||
ColorInfo.ColorFormat.CF_RGB: "RGB",
|
||||
ColorInfo.ColorFormat.CF_YCbCr_422: "YCbCr 4:2:2",
|
||||
ColorInfo.ColorFormat.CF_YCbCr_444: "YCbCr 4:4:4",
|
||||
ColorInfo.ColorFormat.CF_YCbCr_420: "YCbCr 4:2:0",
|
||||
ColorInfo.ColorFormat.CF_Y_ONLY: "Y only",
|
||||
ColorInfo.ColorFormat.CF_RAW: "RAW",
|
||||
ColorInfo.ColorFormat.CF_DSC: "DSC",
|
||||
}
|
||||
fmt = fmt_map.get(
|
||||
ci.color_format,
|
||||
ci.color_format.name.replace("CF_", "").replace("_", " "),
|
||||
)
|
||||
|
||||
if ci.color_format == ColorInfo.ColorFormat.CF_RGB:
|
||||
range_str = (
|
||||
"Legacy RGB mode"
|
||||
if ci.dynamic_range == ColorInfo.DynamicRange.DR_VESA
|
||||
else "Limited range"
|
||||
)
|
||||
elif ci.dynamic_range == ColorInfo.DynamicRange.DR_VESA:
|
||||
range_str = "Full range"
|
||||
else:
|
||||
range_str = "Limited range"
|
||||
return f"{fmt}/{range_str}"
|
||||
|
||||
|
||||
def _max_bitrate_gbps(pg, bpp: int) -> float:
|
||||
caps = pg._caps()
|
||||
effective_bpp = bpp if bpp > 0 else 24
|
||||
return round(caps.max_pix_rate * effective_bpp / 1_000_000_000.0, 3)
|
||||
|
||||
|
||||
def collect_pattern_generator_status(
|
||||
pg,
|
||||
*,
|
||||
pattern_hint: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
vm = pg.get_stream_video_mode()
|
||||
pg_status = pg.status()
|
||||
ci = vm.color_info
|
||||
|
||||
refresh_hz = vm.timing.frame_rate / 1000.0 if vm.timing.frame_rate else 0.0
|
||||
timing = f"{vm.timing.hactive} x {vm.timing.vactive} @ {refresh_hz:g}Hz"
|
||||
current_br = vm.bit_rate
|
||||
max_br = _max_bitrate_gbps(pg, ci.bpp)
|
||||
|
||||
return {
|
||||
"total_bitrate": f"{current_br:.3f} / {max_br:.3f} Gbps",
|
||||
"timing": timing,
|
||||
"pattern": _format_pattern_name(pg, pattern_hint),
|
||||
"cef": _format_cef(ci),
|
||||
"bpc": str(ci.bpc),
|
||||
"status": str(pg_status.error),
|
||||
}
|
||||
|
||||
|
||||
def format_pattern_generator_status_log(status: dict[str, Any]) -> list[str]:
|
||||
return [
|
||||
f"Total bitrate: {status['total_bitrate']}",
|
||||
f" Timing: {status['timing']}",
|
||||
f" Pattern: {status['pattern']}",
|
||||
f" CEF: {status['cef']}",
|
||||
f" BPC: {status['bpc']}",
|
||||
f" Status: {status['status']}",
|
||||
]
|
||||
|
||||
|
||||
def read_pattern_generator_status_lines(
|
||||
pg,
|
||||
*,
|
||||
pattern_hint: str | None = None,
|
||||
) -> list[str]:
|
||||
status = collect_pattern_generator_status(pg, pattern_hint=pattern_hint)
|
||||
return format_pattern_generator_status_log(status)
|
||||
@@ -2,6 +2,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
from app.ucd.domain import (
|
||||
Colorimetry,
|
||||
@@ -33,6 +34,21 @@ class SignalService:
|
||||
def __init__(self, device: IUcdDevice, bus: EventBus):
|
||||
self._dev = device
|
||||
self._bus = bus
|
||||
self._pg_status_log_callback = None
|
||||
|
||||
def set_pg_status_log_callback(self, callback) -> None:
|
||||
"""设置 PG 状态日志回调：``callback(lines: list[str]) -> None``。"""
|
||||
self._pg_status_log_callback = callback
|
||||
|
||||
def _notify_pattern_sent(self, pattern_hint: str | None = None) -> None:
|
||||
if not self._pg_status_log_callback:
|
||||
return
|
||||
try:
|
||||
lines = self.get_pattern_generator_status_lines(pattern_hint)
|
||||
if lines:
|
||||
self._pg_status_log_callback(lines)
|
||||
except Exception:
|
||||
log.exception("SignalService._notify_pattern_sent failed")
|
||||
|
||||
# -- 高层接口 ------------------------------------------------
|
||||
|
||||
@@ -59,17 +75,30 @@ class SignalService:
|
||||
self._dev.apply()
|
||||
return changed
|
||||
|
||||
def send_pattern(self, pattern: PatternSpec) -> None:
|
||||
def send_pattern(
|
||||
self,
|
||||
pattern: PatternSpec,
|
||||
*,
|
||||
pattern_hint: str | None = None,
|
||||
) -> None:
|
||||
"""在已 configure 的信号上仅更新图案后 apply。"""
|
||||
log.info("SignalService.send_pattern pattern=%s", pattern.kind.value)
|
||||
self._dev.set_pattern(pattern)
|
||||
self._dev.apply()
|
||||
self._notify_pattern_sent(pattern_hint)
|
||||
|
||||
def send_solid_rgb(self, rgb: tuple[int, int, int] | list[int]) -> None:
|
||||
self.send_pattern(solid_rgb_pattern(rgb))
|
||||
rgb_list = list(rgb)
|
||||
self.send_pattern(
|
||||
solid_rgb_pattern(rgb_list),
|
||||
pattern_hint=f"Solid Color {rgb_list}",
|
||||
)
|
||||
|
||||
def send_image(self, path: str) -> None:
|
||||
self.send_pattern(image_pattern(path))
|
||||
self.send_pattern(
|
||||
image_pattern(path),
|
||||
pattern_hint=os.path.basename(path) or "Image File",
|
||||
)
|
||||
|
||||
def update_signal_format(
|
||||
self,
|
||||
@@ -127,9 +156,26 @@ class SignalService:
|
||||
"""按 :class:`PQConfig` 写入色彩 / Timing / 当前 Pattern（不 apply 输出）。"""
|
||||
return bool(self._dev.set_ucd_params(config))
|
||||
|
||||
def send_pattern_params(self, params) -> bool:
|
||||
def send_pattern_params(
|
||||
self,
|
||||
params,
|
||||
*,
|
||||
pattern_hint: str | None = None,
|
||||
) -> bool:
|
||||
"""以 ``params`` 更新当前 pattern 的参数并 apply。"""
|
||||
return bool(self._dev.send_current_pattern_params(params))
|
||||
ok = bool(self._dev.send_current_pattern_params(params))
|
||||
if ok:
|
||||
self._notify_pattern_sent(pattern_hint)
|
||||
return ok
|
||||
|
||||
def get_pattern_generator_status_lines(
|
||||
self,
|
||||
pattern_hint: str | None = None,
|
||||
) -> list[str]:
|
||||
getter = getattr(self._dev, "get_pattern_generator_status_lines", None)
|
||||
if not callable(getter):
|
||||
return []
|
||||
return list(getter(pattern_hint) or [])
|
||||
|
||||
def apply_and_run(self, config, pattern_params) -> bool:
|
||||
"""``set_ucd_params`` + ``set_pattern`` + ``run`` 的复合操作。"""
|
||||
@@ -380,7 +426,13 @@ class PatternService:
|
||||
raise IndexError(f"pattern 索引越界: {index}")
|
||||
|
||||
pattern_param = session.pattern_params[index]
|
||||
if not self.app.signal_service.send_pattern_params(pattern_param):
|
||||
pattern_hint = None
|
||||
if session.display_names and index < len(session.display_names):
|
||||
pattern_hint = session.display_names[index]
|
||||
if not self.app.signal_service.send_pattern_params(
|
||||
pattern_param,
|
||||
pattern_hint=pattern_hint,
|
||||
):
|
||||
raise RuntimeError(f"发送 pattern 失败: {index}")
|
||||
return pattern_param
|
||||
|
||||
|
||||
2
app/ucd/solid_color_scale.py
Normal file
2
app/ucd/solid_color_scale.py
Normal file
@@ -0,0 +1,2 @@
|
||||
"""兼容层：请改用 ``app.solid_color_scale``。"""
|
||||
from app.solid_color_scale import * # noqa: F403
|
||||
Reference in New Issue
Block a user