"""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 .api_client import MediaServerClient, MediaServerError from .const import DOMAIN 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.""" client: MediaServerClient = hass.data[DOMAIN][entry.entry_id]["client"] try: monitors = await client.get_display_monitors() except MediaServerError as err: _LOGGER.error("Failed to fetch display monitors: %s", err) return entities: list[Any] = [] for monitor in monitors: if monitor.get("input_source_supported") and monitor.get("available_input_sources"): entities.append(DisplayInputSourceSelect(client, entry, monitor)) if monitor.get("color_preset_supported") and monitor.get("available_color_presets"): entities.append(DisplayColorPresetSelect(client, entry, monitor)) if monitor.get("picture_mode_supported") and monitor.get("available_picture_modes"): entities.append(DisplayPictureModeSelect(client, entry, monitor)) if entities: async_add_entities(entities) _LOGGER.info("Added %d display select entities", len(entities)) class _DisplaySelectBase(SelectEntity): """Shared base for per-display selects.""" _attr_has_entity_name = True def __init__( self, client: MediaServerClient, entry: ConfigEntry, monitor: dict[str, Any], ) -> None: self._client = client self._monitor_id: int = monitor["id"] self._attr_device_info = display_device_info(entry, monitor) class DisplayInputSourceSelect(_DisplaySelectBase): """Switch the monitor's active input (HDMI1, DP1, ...).""" _attr_name = "Input source" _attr_icon = "mdi:video-input-hdmi" def __init__( self, client: MediaServerClient, entry: ConfigEntry, monitor: dict[str, Any], ) -> None: super().__init__(client, entry, monitor) self._attr_unique_id = f"{entry.entry_id}_display_input_{self._monitor_id}" self._attr_options = list(monitor.get("available_input_sources") or []) current = monitor.get("input_source") self._attr_current_option = 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, ) # Re-read so the entity state reflects what the monitor actually did. await self.async_update() self.async_write_ha_state() return self._attr_current_option = option self.async_write_ha_state() async def async_update(self) -> None: try: monitors = await self._client.get_display_monitors() for monitor in monitors: if monitor["id"] == self._monitor_id: current = monitor.get("input_source") self._attr_current_option = current if current in self._attr_options else None break except MediaServerError as err: _LOGGER.error("Failed to refresh input source for monitor %d: %s", self._monitor_id, err) class DisplayColorPresetSelect(_DisplaySelectBase): """Switch the monitor's color temperature preset (sRGB / 6500K / ...).""" _attr_name = "Color preset" _attr_icon = "mdi:palette" def __init__( self, client: MediaServerClient, entry: ConfigEntry, monitor: dict[str, Any], ) -> None: super().__init__(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 []) current = monitor.get("color_preset") self._attr_current_option = 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.async_update() self.async_write_ha_state() return self._attr_current_option = option self.async_write_ha_state() async def async_update(self) -> None: try: monitors = await self._client.get_display_monitors() for monitor in monitors: if monitor["id"] == self._monitor_id: current = monitor.get("color_preset") self._attr_current_option = current if current in self._attr_options else None break except MediaServerError as err: _LOGGER.error("Failed to refresh color preset for monitor %d: %s", self._monitor_id, err) class DisplayPictureModeSelect(_DisplaySelectBase): """Switch the monitor's picture/scene mode via VCP 0xDC. The server returns options as `[{code: int, label: str}, ...]`. We use labels as the user-facing options and keep a label→code map for writes. """ _attr_name = "Picture mode" _attr_icon = "mdi:image-multiple" def __init__( self, client: MediaServerClient, entry: ConfigEntry, monitor: dict[str, Any], ) -> None: super().__init__(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()) current = monitor.get("picture_mode") self._attr_current_option = 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.async_update() self.async_write_ha_state() return self._attr_current_option = option self.async_write_ha_state() async def async_update(self) -> None: try: monitors = await self._client.get_display_monitors() for monitor in monitors: if monitor["id"] == self._monitor_id: current = monitor.get("picture_mode") self._attr_current_option = current if current in self._attr_options else None break except MediaServerError as err: _LOGGER.error("Failed to refresh picture mode for monitor %d: %s", self._monitor_id, err)