pqAutomationApp/drivers/ucd_driver.py

"""UCD 驱动层。

唯一暴露给上层的入口为 :class:`IUcdDevice` 抽象接口，以及实现：
:class:`UCD323Device`（生产）和 :class:`FakeUcdDevice`（单测）。

Phase 1 实现策略
-----------------
为保证零行为变更，:class:`UCD323Device` 当前**内部委托**给已有的
:class:`drivers.UCD323_Function.UCDController`。后续 Phase 2 会将
SDK 调用直接搬入本模块，届时可删除旧 ``UCDController`` 文件。

文件分区：
    §1 DeviceInfo / list_devices
    §2 IUcdDevice 抽象接口
    §3 UCD323Device 真实实现
    §4 FakeUcdDevice 单测实现
"""

from __future__ import annotations

from contextlib import contextmanager
import logging
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import TYPE_CHECKING

from app.ucd_domain import (
    ConnectionChanged,
    EventBus,
    Interface,
    PatternApplied,
    PatternKind,
    PatternSpec,
    SignalApplied,
    SignalFormat,
    TimingSpec,
    UcdApplyFailed,
    UcdConfigError,
    UcdNotConnected,
    UcdSdkError,
    UcdState,
    UcdStateError,
    assert_transition,
)

if TYPE_CHECKING:
    from drivers.UCD323_Function import UCDController

log = logging.getLogger(__name__)

_DEVICE_LOCK_TIMEOUT_SECONDS = 8.0


# ─── §1 DeviceInfo / list_devices ────────────────────────────────


@dataclass(frozen=True)
class DeviceInfo:
    """UCD 设备发现条目。

    ``display`` 是 SDK 给出的完整字符串（``"0: UCD-323 #12345678"``）；
    ``index`` / ``serial`` / ``model`` 通过解析得到，解析失败时为 None。
    """

    display: str
    index: int | None = None
    serial: str | None = None
    model: str | None = None

    @classmethod
    def parse(cls, display: str) -> "DeviceInfo":
        idx: int | None = None
        model: str | None = None
        serial: str | None = None
        try:
            head, rest = display.split(":", 1)
            idx = int(head.strip())
            rest = rest.strip()
            # 形如 "UCD-323 #12345678" 或 "UCD-323 #12345678 (in use)"
            tokens = rest.split()
            if tokens:
                model = tokens[0]
            for tok in tokens[1:]:
                if tok.startswith("#") and len(tok) >= 2:
                    serial = tok.lstrip("#")
                    break
        except Exception:  # noqa: BLE001 - 解析失败保留原 display 即可
            pass
        return cls(display=display, index=idx, serial=serial, model=model)


def list_devices(controller: "UCDController") -> list[DeviceInfo]:
    """通过给定的底层 controller 枚举可用 UCD 设备。"""
    try:
        raw_list = controller.search_device()
    except Exception as exc:  # noqa: BLE001
        raise UcdSdkError("枚举 UCD 设备失败") from exc
    return [DeviceInfo.parse(s) for s in (raw_list or [])]


# ─── §2 IUcdDevice 抽象接口 ──────────────────────────────────────


class IUcdDevice(ABC):
    """UCD 信号发生器抽象设备。

    上层（Service / GUI）**只**通过本接口操作硬件，不得穿透到
    UniTAP SDK 或具体实现细节。
    """

    @property
    @abstractmethod
    def state(self) -> UcdState: ...

    @property
    @abstractmethod
    def info(self) -> DeviceInfo | None: ...

    @abstractmethod
    def open(self, info: DeviceInfo, *, interface: Interface = Interface.HDMI) -> None:
        """打开设备并选择接口角色。失败抛 :class:`UcdSdkError` 等。"""

    @abstractmethod
    def close(self) -> None:
        """关闭设备（幂等）。"""

    @abstractmethod
    def configure(self, signal: SignalFormat, timing: TimingSpec) -> bool:
        """写入信号格式与 timing（未 apply）。返回 ``format_changed``。"""

    @abstractmethod
    def set_pattern(self, pattern: PatternSpec) -> None:
        """设置当前图案（未 apply）。"""

    @abstractmethod
    def apply(self) -> None:
        """将已配置的信号格式 + 图案一次性提交给硬件。"""

    @abstractmethod
    def current_resolution(self) -> tuple[int, int]:
        """读取当前 timing 的 (width, height)；未连接时返回默认 (3840, 2160)。"""


# ─── §3 UCD323Device 真实实现 ────────────────────────────────────


class UCD323Device(IUcdDevice):
    """生产环境实现。内部委托给传统 :class:`UCDController`（Phase 1）。"""

    def __init__(self, bus: EventBus, controller: "UCDController | None" = None):
        from drivers.UCD323_Function import UCDController as _UCDController

        self._bus = bus
        self._controller: "UCDController" = controller or _UCDController()
        self._state: UcdState = UcdState.CLOSED
        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
    def state(self) -> UcdState:
        return self._state

    @property
    def info(self) -> DeviceInfo | None:
        return self._info

    @property
    def raw_controller(self) -> "UCDController":
        """Phase 1 过渡期：给暂未迁移的旧调用点的逃生通道。

        新代码**不**应使用本属性，迁移完成后即可删除。
        """
        return self._controller

    # -- 生命周期 ------------------------------------------------

    def open(self, info: DeviceInfo, *, interface: Interface = Interface.HDMI) -> None:
        with self._acquire_device_lock("open"):
            assert_transition(self._state, UcdState.OPENED)
            if interface is not Interface.HDMI:
                # Phase 1：底层 UCDController.open() 写死了 HDMISource。
                raise UcdConfigError(
                    f"暂不支持接口 {interface.value}；当前仅实现 HDMI"
                )
            try:
                ok = self._controller.open(info.display)
            except Exception as exc:  # noqa: BLE001
                raise UcdSdkError(f"打开设备失败: {info.display}") from exc
            if not ok:
                raise UcdSdkError(f"打开设备失败: {info.display}")
            self._info = info
            self._interface = interface
            self._state = UcdState.OPENED
            self._bus.publish(ConnectionChanged(True, info.serial))

    def close(self) -> None:
        with self._acquire_device_lock("close"):
            if self._state == UcdState.CLOSED:
                return
            try:
                self._controller.close()
            except Exception:  # noqa: BLE001
                log.exception("关闭 UCD 时发生异常")
            self._state = UcdState.CLOSED
            self._curr_signal = None
            self._curr_timing = None
            self._curr_pattern = None
            self._last_applied = None
            prev_serial = self._info.serial if self._info else None
            self._info = None
            self._bus.publish(ConnectionChanged(False, prev_serial))

    # -- 配置 ----------------------------------------------------

    def configure(self, signal: SignalFormat, timing: TimingSpec) -> bool:
        with self._acquire_device_lock("configure"):
            if self._state == UcdState.CLOSED:
                raise UcdNotConnected("UCD 未连接，无法 configure")
            try:
                # 颜色模式（color_format / bpc / colorimetry）
                if not self._controller.set_color_mode(
                    signal.color_format.value,
                    int(signal.bpc),
                    _colorimetry_to_legacy_key(signal),
                ):
                    raise UcdConfigError(
                        f"set_color_mode 失败: {signal!r}"
                    )
                # dynamic_range 在新接口中是一等公民
                self._apply_dynamic_range(signal)

                # Timing
                if not self._controller.set_timing_from_string(str(timing)):
                    raise UcdConfigError(f"set_timing_from_string 失败: {timing}")
            except UcdConfigError:
                raise
            except Exception as exc:  # noqa: BLE001
                raise UcdSdkError("configure 异常") from exc

            self._curr_signal = signal
            self._curr_timing = timing
            self._state = UcdState.CONFIGURED
            return (signal, timing) != self._last_applied

    def set_pattern(self, pattern: PatternSpec) -> None:
        with self._acquire_device_lock("set_pattern"):
            # Phase 2 过渡：允许从 OPENED 直接 set_pattern——遗留路径
            # （test_runner 等）通过旧 controller.apply_signal_format 写入
            # 信号格式，未经过本设备的 configure。此时 self._state 仍为
            # OPENED，但硬件实际已处于可接收 pattern 状态。
            if self._state == UcdState.CLOSED:
                raise UcdNotConnected("UCD 未连接，无法 set_pattern")
            self._curr_pattern = pattern
            # 仅本地暂存，真正写硬件在 apply()

    def apply(self) -> None:
        with self._acquire_device_lock("apply"):
            if self._state == UcdState.CLOSED:
                raise UcdNotConnected("UCD 未连接，无法 apply")
            if self._curr_pattern is None:
                raise UcdStateError("apply 前必须先 set_pattern")
            try:
                ok = self._apply_pattern_via_controller(self._curr_pattern)
            except Exception as exc:  # noqa: BLE001
                raise UcdSdkError(
                    f"apply 异常: {type(exc).__name__}: {exc}"
                ) from exc
            if not ok:
                raise UcdApplyFailed(
                    f"apply 失败: pattern={self._curr_pattern.kind.value}"
                )

            # SignalApplied 事件仅在通过新 API configure 过时发出；
            # 遗留路径下 self._curr_signal/_curr_timing 可能为 None。
            if self._curr_signal is not None and self._curr_timing is not None:
                changed = (self._curr_signal, self._curr_timing) != self._last_applied
                self._last_applied = (self._curr_signal, self._curr_timing)
                self._bus.publish(
                    SignalApplied(self._curr_signal, self._curr_timing, changed)
                )
            self._state = UcdState.APPLIED
            self._bus.publish(PatternApplied(self._curr_pattern))

    # -- 查询 ----------------------------------------------------

    def current_resolution(self) -> tuple[int, int]:
        try:
            return self._controller.get_current_resolution((3840, 2160))
        except Exception:  # noqa: BLE001
            return (3840, 2160)

    # -- 内部辅助 ------------------------------------------------

    def _apply_dynamic_range(self, signal: SignalFormat) -> None:
        import UniTAP  # 局部导入，避免本模块在无 SDK 环境下导入即失败

        from app.ucd_domain import DynamicRange

        ci = self._controller.color_info
        if ci is None:
            return
        if signal.dynamic_range is DynamicRange.FULL:
            ci.dynamic_range = UniTAP.ColorInfo.DynamicRange.DR_VESA
        else:
            ci.dynamic_range = UniTAP.ColorInfo.DynamicRange.DR_CTA

    def _apply_pattern_via_controller(self, pattern: PatternSpec) -> bool:
        """根据 PatternKind 走最合适的旧 controller 路径。"""
        if pattern.kind is PatternKind.IMAGE:
            if not pattern.image_path:
                raise UcdConfigError("IMAGE pattern 必须提供 image_path")
            return bool(self._controller.send_image_pattern(pattern.image_path))

        # 预定义图案路径：复用 controller.set_pattern + run()
        from drivers.UCD323_Enum import UCDEnum  # 局部导入避免循环

        video_pattern = UCDEnum.VideoPatternInfo.get_video_pattern(pattern.kind.value)
        if video_pattern is None:
            raise UcdConfigError(f"不支持的 PatternKind: {pattern.kind!r}")
        self._controller.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:
            params = list(pattern.extras)

        if not self._controller.set_pattern(video_pattern, params):
            raise UcdApplyFailed("controller.set_pattern 返回 False")
        # 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:
    """新 :class:`Colorimetry` → 旧 ``UCDEnum.ColorInfo.get_colorimetry`` 的 key。

    BT.2020 在 YCbCr / RGB 输出下走不同 SDK 枚举（参考旧
    ``_get_colorimetry_from_color_space`` 的逻辑），这里也做同样的分支。
    """
    from app.ucd_domain import Colorimetry, is_ycbcr

    cm = signal.colorimetry
    ycbcr = is_ycbcr(signal.color_format)

    if cm is Colorimetry.BT2020:
        return "bt2020ycbcr" if ycbcr else "bt2020rgb"
    return {
        Colorimetry.SRGB: "srgb",
        Colorimetry.BT709: "bt709",
        Colorimetry.BT601: "bt601",
        Colorimetry.DCI_P3: "dcip3",
        Colorimetry.ADOBE_RGB: "adobergb",
    }.get(cm, "srgb")


# ─── §4 FakeUcdDevice 单测实现 ───────────────────────────────────


class FakeUcdDevice(IUcdDevice):
    """无硬件依赖的 Fake 实现；记录调用序列供单测断言。"""

    def __init__(self, bus: EventBus | None = None) -> None:
        self._bus = bus or EventBus()
        self._state = UcdState.CLOSED
        self._info: DeviceInfo | None = None
        self._signal: SignalFormat | None = None
        self._timing: TimingSpec | None = None
        self._pattern: PatternSpec | None = None
        self._last_applied: tuple[SignalFormat, TimingSpec] | None = None
        self.calls: list[tuple] = []  # ("open", info) / ("configure", ...) ...

    @property
    def state(self) -> UcdState:
        return self._state

    @property
    def info(self) -> DeviceInfo | None:
        return self._info

    def open(self, info: DeviceInfo, *, interface: Interface = Interface.HDMI) -> None:
        assert_transition(self._state, UcdState.OPENED)
        self.calls.append(("open", info, interface))
        self._info = info
        self._state = UcdState.OPENED
        self._bus.publish(ConnectionChanged(True, info.serial))

    def close(self) -> None:
        if self._state == UcdState.CLOSED:
            return
        self.calls.append(("close",))
        self._state = UcdState.CLOSED
        prev = self._info.serial if self._info else None
        self._info = None
        self._signal = self._timing = self._pattern = None
        self._last_applied = None
        self._bus.publish(ConnectionChanged(False, prev))

    def configure(self, signal: SignalFormat, timing: TimingSpec) -> bool:
        if self._state == UcdState.CLOSED:
            raise UcdNotConnected()
        self.calls.append(("configure", signal, timing))
        self._signal, self._timing = signal, timing
        self._state = UcdState.CONFIGURED
        return (signal, timing) != self._last_applied

    def set_pattern(self, pattern: PatternSpec) -> None:
        if self._state not in (UcdState.CONFIGURED, UcdState.APPLIED):
            raise UcdStateError(f"非法状态 {self._state.name}")
        self.calls.append(("set_pattern", pattern))
        self._pattern = pattern

    def apply(self) -> None:
        if self._signal is None or self._timing is None:
            raise UcdStateError("apply 前必须 configure")
        if self._pattern is None:
            raise UcdStateError("apply 前必须 set_pattern")
        self.calls.append(("apply",))
        changed = (self._signal, self._timing) != self._last_applied
        self._last_applied = (self._signal, self._timing)
        self._state = UcdState.APPLIED
        self._bus.publish(SignalApplied(self._signal, self._timing, changed))
        self._bus.publish(PatternApplied(self._pattern))

    def current_resolution(self) -> tuple[int, int]:
        if self._timing is None:
            return (3840, 2160)
        return (self._timing.width, self._timing.height)


__all__ = [
    "DeviceInfo",
    "list_devices",
    "IUcdDevice",
    "UCD323Device",
    "FakeUcdDevice",
]