haos-hacs-integration-media-player/custom_components/remote_media_player/select.py

"""Select platform: DDC/CI input source, color preset, and picture mode."""

from __future__ import annotations

import logging
from typing import Any

from homeassistant.components.select import SelectEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity

from .api_client import MediaServerClient, MediaServerError
from .const import DOMAIN
from .display_coordinator import DisplayCoordinator
from .display_device import display_device_info

_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up per-display select entities."""
    data = hass.data[DOMAIN][entry.entry_id]
    client: MediaServerClient = data["client"]
    coordinator: DisplayCoordinator = data["display_coordinator"]

    if not coordinator.data:
        return

    entities: list[Any] = []
    for monitor in coordinator.data.values():
        if monitor.get("input_source_supported") and monitor.get("available_input_sources"):
            entities.append(DisplayInputSourceSelect(coordinator, client, entry, monitor))
        if monitor.get("color_preset_supported") and monitor.get("available_color_presets"):
            entities.append(DisplayColorPresetSelect(coordinator, client, entry, monitor))
        if monitor.get("picture_mode_supported") and monitor.get("available_picture_modes"):
            entities.append(DisplayPictureModeSelect(coordinator, client, entry, monitor))

    if entities:
        async_add_entities(entities)
        _LOGGER.info("Added %d display select entities", len(entities))


class _DisplaySelectBase(CoordinatorEntity[DisplayCoordinator], SelectEntity):
    """Shared base for per-display selects."""

    _attr_has_entity_name = True

    def __init__(
        self,
        coordinator: DisplayCoordinator,
        client: MediaServerClient,
        entry: ConfigEntry,
        monitor: dict[str, Any],
    ) -> None:
        super().__init__(coordinator)
        self._client = client
        self._monitor_id: int = monitor["id"]
        self._attr_device_info = display_device_info(entry, monitor)

    @property
    def _monitor(self) -> dict[str, Any]:
        if self.coordinator.data is None:
            return {}
        return self.coordinator.data.get(self._monitor_id, {})


class DisplayInputSourceSelect(_DisplaySelectBase):
    """Switch the monitor's active input (HDMI1, DP1, ...)."""

    _attr_name = "Input source"
    _attr_icon = "mdi:video-input-hdmi"

    def __init__(
        self,
        coordinator: DisplayCoordinator,
        client: MediaServerClient,
        entry: ConfigEntry,
        monitor: dict[str, Any],
    ) -> None:
        super().__init__(coordinator, client, entry, monitor)
        self._attr_unique_id = f"{entry.entry_id}_display_input_{self._monitor_id}"
        # Available inputs are a static EDID/DDC capability — capturing them
        # at discovery avoids re-allocating the option list on every poll.
        self._attr_options = list(monitor.get("available_input_sources") or [])

    @property
    def current_option(self) -> str | None:
        current = self._monitor.get("input_source")
        return current if current in self._attr_options else None

    async def async_select_option(self, option: str) -> None:
        try:
            result = await self._client.set_display_input_source(self._monitor_id, option)
        except MediaServerError as err:
            _LOGGER.error("Failed to set input source for monitor %d: %s", self._monitor_id, err)
            return
        if not result.get("success"):
            _LOGGER.warning(
                "Monitor %d rejected input source %s (DDC/CI silently dropped)",
                self._monitor_id, option,
            )
            await self.coordinator.async_request_refresh()
            return
        self.coordinator.apply_optimistic(self._monitor_id, input_source=option)


class DisplayColorPresetSelect(_DisplaySelectBase):
    """Switch the monitor's color temperature preset (sRGB / 6500K / ...)."""

    _attr_name = "Color preset"
    _attr_icon = "mdi:palette"

    def __init__(
        self,
        coordinator: DisplayCoordinator,
        client: MediaServerClient,
        entry: ConfigEntry,
        monitor: dict[str, Any],
    ) -> None:
        super().__init__(coordinator, client, entry, monitor)
        self._attr_unique_id = f"{entry.entry_id}_display_color_preset_{self._monitor_id}"
        self._attr_options = list(monitor.get("available_color_presets") or [])

    @property
    def current_option(self) -> str | None:
        current = self._monitor.get("color_preset")
        return current if current in self._attr_options else None

    async def async_select_option(self, option: str) -> None:
        try:
            result = await self._client.set_display_color_preset(self._monitor_id, option)
        except MediaServerError as err:
            _LOGGER.error("Failed to set color preset for monitor %d: %s", self._monitor_id, err)
            return
        if not result.get("success"):
            _LOGGER.warning(
                "Monitor %d rejected color preset %s (DDC/CI silently dropped)",
                self._monitor_id, option,
            )
            await self.coordinator.async_request_refresh()
            return
        self.coordinator.apply_optimistic(self._monitor_id, color_preset=option)


class DisplayPictureModeSelect(_DisplaySelectBase):
    """Switch the monitor's picture/scene mode via VCP 0xDC.

    The server returns options as ``[{code: int, label: str}, ...]``. Labels
    are exposed as user-facing options and a label→code map drives writes.
    """

    _attr_name = "Picture mode"
    _attr_icon = "mdi:image-multiple"

    def __init__(
        self,
        coordinator: DisplayCoordinator,
        client: MediaServerClient,
        entry: ConfigEntry,
        monitor: dict[str, Any],
    ) -> None:
        super().__init__(coordinator, client, entry, monitor)
        self._attr_unique_id = f"{entry.entry_id}_display_picture_mode_{self._monitor_id}"

        modes = monitor.get("available_picture_modes") or []
        self._label_to_code: dict[str, int] = {
            mode["label"]: mode["code"]
            for mode in modes
            if "label" in mode and "code" in mode
        }
        self._attr_options = list(self._label_to_code.keys())

    @property
    def current_option(self) -> str | None:
        current = self._monitor.get("picture_mode")
        return current if current in self._attr_options else None

    async def async_select_option(self, option: str) -> None:
        code = self._label_to_code.get(option)
        if code is None:
            _LOGGER.error("Unknown picture mode label: %s", option)
            return
        try:
            result = await self._client.set_display_picture_mode(self._monitor_id, code)
        except MediaServerError as err:
            _LOGGER.error("Failed to set picture mode for monitor %d: %s", self._monitor_id, err)
            return
        if not result.get("success"):
            _LOGGER.warning(
                "Monitor %d rejected picture mode %s (code %d) - monitor's DDC/CI"
                " implementation of VCP 0xDC may be incomplete",
                self._monitor_id, option, code,
            )
            await self.coordinator.async_request_refresh()
            return
        self.coordinator.apply_optimistic(self._monitor_id, picture_mode=option)