"""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_translation_key = "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_translation_key = "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_translation_key = "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)