pqAutomationApp/app/services/ucd_service.py

"""UCD 信号 / 图案应用服务层。

服务层是 GUI ↔ Driver 的唯一通道，负责：
- 将 UI 字符串（"BT.709"、"10bit"、"YCbCr 4:4:4" 等）翻译成 :class:`SignalFormat`；
- 将各 panel 的 timing 字符串翻译成 :class:`TimingSpec`；
- 协调 :meth:`IUcdDevice.configure` / ``set_pattern`` / ``apply`` 的调用顺序；
- 通过 :class:`EventBus` 让 GUI 订阅状态变化，而非主动轮询。

本层不直接 import UniTAP，也不读取 :mod:`tkinter` 变量；
所有输入都是显式参数，便于单测。
"""

from __future__ import annotations

import logging
import threading

from app.ucd_domain import (
    Colorimetry,
    DynamicRange,
    EventBus,
    PatternKind,
    PatternSpec,
    SignalFormat,
    TimingSpec,
    UcdError,
    bit_depth_str_to_bpc,
    color_space_to_colorimetry,
    data_range_to_dynamic_range,
    output_format_to_color_format,
    parse_timing_str,
)
from drivers.ucd_driver import IUcdDevice

log = logging.getLogger(__name__)


# ─── 视图字符串 → 值对象 转换工具 ────────────────────────────────


def build_signal_format(
    *,
    color_space: str,
    output_format: str,
    bit_depth: str,
    data_range: str = "Full",
) -> SignalFormat:
    """根据下拉框字符串组装 :class:`SignalFormat`。

    各参数解析失败抛 :class:`UcdConfigError`。
    """
    return SignalFormat(
        color_format=output_format_to_color_format(output_format),
        colorimetry=color_space_to_colorimetry(color_space),
        bpc=bit_depth_str_to_bpc(bit_depth),
        dynamic_range=data_range_to_dynamic_range(data_range),
    )


def build_timing(timing_str: str) -> TimingSpec:
    """``"DMT 3840x2160@60Hz"`` → :class:`TimingSpec`。"""
    return parse_timing_str(timing_str)


def solid_rgb_pattern(rgb: tuple[int, int, int] | list[int]) -> PatternSpec:
    r, g, b = rgb[0], rgb[1], rgb[2]
    return PatternSpec(kind=PatternKind.SOLID, solid_rgb=(int(r), int(g), int(b)))


def image_pattern(path: str) -> PatternSpec:
    return PatternSpec(kind=PatternKind.IMAGE, image_path=path)


# ─── 服务 ────────────────────────────────────────────────────────


class SignalService:
    """协调 SignalFormat / Timing / Pattern 的写入与提交。

    使用线程锁串行化所有对外的 ``apply_*`` 调用，避免多个测试线程
    同时操作 UCD 造成 SDK 状态错乱。
    """

    def __init__(self, device: IUcdDevice, bus: EventBus):
        self._dev = device
        self._bus = bus
        self._lock = threading.RLock()

    # -- 高层接口 ------------------------------------------------

    def apply(
        self,
        *,
        signal: SignalFormat,
        timing: TimingSpec,
        pattern: PatternSpec,
    ) -> bool:
        """一次性提交信号格式 + timing + 图案。

        Returns:
            ``format_changed``——本次相对上一次 :meth:`apply` 是否变化。
        """
        with self._lock:
            log.info(
                "SignalService.apply signal=%s timing=%s pattern=%s",
                signal,
                timing,
                pattern.kind.value,
            )
            changed = self._dev.configure(signal, timing)
            self._dev.set_pattern(pattern)
            self._dev.apply()
            return changed

    def send_pattern(self, pattern: PatternSpec) -> None:
        """在已 configure 的信号上仅更新图案后 apply。"""
        with self._lock:
            log.info("SignalService.send_pattern pattern=%s", pattern.kind.value)
            self._dev.set_pattern(pattern)
            self._dev.apply()

    def send_solid_rgb(self, rgb: tuple[int, int, int] | list[int]) -> None:
        self.send_pattern(solid_rgb_pattern(rgb))

    def send_image(self, path: str) -> None:
        self.send_pattern(image_pattern(path))

    # -- 过渡期 API（Phase 2）-----------------------------------
    # 现有 GUI 回调以"仅更新信号格式、不切换图案"的方式调用
    # ``ucd.apply_signal_format(color_space=..., color_format=..., bit_depth=...)``。
    # 新代码统一通过本方法走 SignalService；内部仍委托给底层
    # controller 的同名旧接口，迁移完成后将替换为纯净实现。

    def update_signal_format(
        self,
        *,
        color_space: str,
        output_format: str,
        bit_depth: str,
        data_range: str = "Full",
        max_cll: int | None = None,
        max_fall: int | None = None,
    ) -> bool:
        """仅将信号格式 stage 到 SDK（沿用上一次的 timing），不切换图案。

        UI 字符串先经域层解析做参数校验；解析失败抛 :class:`UcdConfigError`。
        """
        # 解析仅做校验；当前实现走 raw controller 的旧 API
        _ = build_signal_format(
            color_space=color_space,
            output_format=output_format,
            bit_depth=bit_depth,
            data_range=data_range,
        )
        ctrl = getattr(self._dev, "raw_controller", None)
        if ctrl is None:
            raise UcdError("update_signal_format 暂仅支持 UCD323Device")
        with self._lock:
            return bool(
                ctrl.apply_signal_format(
                    color_space=color_space,
                    color_format=output_format,
                    bit_depth=bit_depth,
                    data_range=data_range,
                    max_cll=max_cll,
                    max_fall=max_fall,
                )
            )

    # -- 透传给上层的查询 ---------------------------------------

    @property
    def device(self) -> IUcdDevice:
        return self._dev

    def current_resolution(self) -> tuple[int, int]:
        return self._dev.current_resolution()


__all__ = [
    "SignalService",
    "build_signal_format",
    "build_timing",
    "solid_rgb_pattern",
    "image_pattern",
    # 重导出常用域类型方便上层 import 一次到位
    "SignalFormat",
    "TimingSpec",
    "PatternSpec",
    "PatternKind",
    "Colorimetry",
    "DynamicRange",
    "UcdError",
]