From 2562894265cc470bb0db3d3c065416764f21342b Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Fri, 27 Feb 2026 19:29:44 -0500 Subject: [PATCH] add denon rs232 integration --- CODEOWNERS | 2 + homeassistant/brands/denon.json | 2 +- .../components/denon_rs232/__init__.py | 56 +++ .../components/denon_rs232/config_flow.py | 142 ++++++++ homeassistant/components/denon_rs232/const.py | 12 + .../components/denon_rs232/manifest.json | 13 + .../components/denon_rs232/media_player.py | 235 +++++++++++++ .../components/denon_rs232/quality_scale.yaml | 60 ++++ .../components/denon_rs232/strings.json | 86 +++++ homeassistant/generated/config_flows.py | 1 + homeassistant/generated/integrations.json | 6 + requirements_all.txt | 3 + requirements_test_all.txt | 3 + tests/components/denon_rs232/__init__.py | 4 + tests/components/denon_rs232/conftest.py | 194 +++++++++++ .../denon_rs232/test_config_flow.py | 244 ++++++++++++++ .../denon_rs232/test_media_player.py | 318 ++++++++++++++++++ 17 files changed, 1380 insertions(+), 1 deletion(-) create mode 100644 homeassistant/components/denon_rs232/__init__.py create mode 100644 homeassistant/components/denon_rs232/config_flow.py create mode 100644 homeassistant/components/denon_rs232/const.py create mode 100644 homeassistant/components/denon_rs232/manifest.json create mode 100644 homeassistant/components/denon_rs232/media_player.py create mode 100644 homeassistant/components/denon_rs232/quality_scale.yaml create mode 100644 homeassistant/components/denon_rs232/strings.json create mode 100644 tests/components/denon_rs232/__init__.py create mode 100644 tests/components/denon_rs232/conftest.py create mode 100644 tests/components/denon_rs232/test_config_flow.py create mode 100644 tests/components/denon_rs232/test_media_player.py diff --git a/CODEOWNERS b/CODEOWNERS index 32705e6c684..6cf9cf1aba7 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -355,6 +355,8 @@ build.json @home-assistant/supervisor /tests/components/deluge/ @tkdrob /homeassistant/components/demo/ @home-assistant/core /tests/components/demo/ @home-assistant/core +/homeassistant/components/denon_rs232/ @balloob +/tests/components/denon_rs232/ @balloob /homeassistant/components/denonavr/ @ol-iver @starkillerOG /tests/components/denonavr/ @ol-iver @starkillerOG /homeassistant/components/derivative/ @afaucogney @karwosts diff --git a/homeassistant/brands/denon.json b/homeassistant/brands/denon.json index a60750e1a31..d4f68c1306e 100644 --- a/homeassistant/brands/denon.json +++ b/homeassistant/brands/denon.json @@ -1,5 +1,5 @@ { "domain": "denon", "name": "Denon", - "integrations": ["denon", "denonavr", "heos"] + "integrations": ["denon", "denonavr", "denon_rs232", "heos"] } diff --git a/homeassistant/components/denon_rs232/__init__.py b/homeassistant/components/denon_rs232/__init__.py new file mode 100644 index 00000000000..820daf3d9d9 --- /dev/null +++ b/homeassistant/components/denon_rs232/__init__.py @@ -0,0 +1,56 @@ +"""The Denon RS232 integration.""" + +from __future__ import annotations + +from denon_rs232 import DenonReceiver, ReceiverState +from denon_rs232.models import MODELS + +from homeassistant.const import CONF_DEVICE, CONF_MODEL, Platform +from homeassistant.core import HomeAssistant, callback +from homeassistant.exceptions import ConfigEntryNotReady + +from .const import ( + DOMAIN, # noqa: F401 + LOGGER, + DenonRS232ConfigEntry, +) + +PLATFORMS = [Platform.MEDIA_PLAYER] + + +async def async_setup_entry(hass: HomeAssistant, entry: DenonRS232ConfigEntry) -> bool: + """Set up Denon RS232 from a config entry.""" + port = entry.data[CONF_DEVICE] + model = MODELS[entry.data[CONF_MODEL]] + receiver = DenonReceiver(port, model=model) + + try: + await receiver.connect() + await receiver.query_state() + except (ConnectionError, OSError) as err: + LOGGER.error("Error connecting to Denon receiver at %s: %s", port, err) + raise ConfigEntryNotReady from err + + entry.runtime_data = receiver + + @callback + def _on_disconnect(state: ReceiverState | None) -> None: + if state is None: + LOGGER.warning("Denon receiver disconnected, reloading config entry") + hass.config_entries.async_schedule_reload(entry.entry_id) + + entry.async_on_unload(receiver.subscribe(_on_disconnect)) + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: DenonRS232ConfigEntry) -> bool: + """Unload a config entry.""" + unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + + if unload_ok: + await entry.runtime_data.disconnect() + + return unload_ok diff --git a/homeassistant/components/denon_rs232/config_flow.py b/homeassistant/components/denon_rs232/config_flow.py new file mode 100644 index 00000000000..2365ca14ee8 --- /dev/null +++ b/homeassistant/components/denon_rs232/config_flow.py @@ -0,0 +1,142 @@ +"""Config flow for the Denon RS232 integration.""" + +from __future__ import annotations + +import os +from typing import Any + +from denon_rs232 import DenonReceiver +from denon_rs232.models import MODELS +import voluptuous as vol + +from homeassistant.components.usb import human_readable_device_name, scan_serial_ports +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult +from homeassistant.const import CONF_DEVICE, CONF_MODEL + +from .const import DOMAIN, LOGGER + +MODEL_OPTIONS = {key: model.name for key, model in MODELS.items()} + +OPTION_PICK_MANUAL = "Enter Manually" + + +async def _async_attempt_connect(port: str, model_key: str) -> str | None: + """Attempt to connect to the receiver at the given port. + + Returns None on success, error on failure. + """ + model = MODELS[model_key] + receiver = DenonReceiver(port, model=model) + + try: + await receiver.connect() + except ( + # When the port contains invalid connection data + ValueError, + # If it is a remote port, and we cannot connect + ConnectionError, + OSError, + ): + return "cannot_connect" + except Exception: # noqa: BLE001 + LOGGER.exception("Unexpected exception") + return "unknown" + else: + await receiver.disconnect() + return None + + +class DenonRS232ConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for Denon RS232.""" + + VERSION = 1 + + _model: str + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle the initial step.""" + errors: dict[str, str] = {} + + if user_input is not None: + if user_input[CONF_DEVICE] == OPTION_PICK_MANUAL: + self._model = user_input[CONF_MODEL] + return await self.async_step_manual() + + self._async_abort_entries_match({CONF_DEVICE: user_input[CONF_DEVICE]}) + error = await _async_attempt_connect( + user_input[CONF_DEVICE], user_input[CONF_MODEL] + ) + if not error: + return self.async_create_entry( + title=MODELS[user_input[CONF_MODEL]].name, + data={ + CONF_DEVICE: user_input[CONF_DEVICE], + CONF_MODEL: user_input[CONF_MODEL], + }, + ) + errors["base"] = error + + ports = await self.hass.async_add_executor_job(get_ports) + ports[OPTION_PICK_MANUAL] = OPTION_PICK_MANUAL + + if user_input is None and ports: + user_input = {CONF_DEVICE: next(iter(ports))} + + return self.async_show_form( + step_id="user", + data_schema=self.add_suggested_values_to_schema( + vol.Schema( + { + vol.Required(CONF_MODEL): vol.In(MODEL_OPTIONS), + vol.Required(CONF_DEVICE): vol.In(ports), + } + ), + user_input or {}, + ), + errors=errors, + ) + + async def async_step_manual( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle a manual port selection.""" + errors: dict[str, str] = {} + + if user_input is not None: + self._async_abort_entries_match({CONF_DEVICE: user_input[CONF_DEVICE]}) + error = await _async_attempt_connect(user_input[CONF_DEVICE], self._model) + if not error: + return self.async_create_entry( + title=MODELS[self._model].name, + data={ + CONF_DEVICE: user_input[CONF_DEVICE], + CONF_MODEL: self._model, + }, + ) + errors["base"] = error + + return self.async_show_form( + step_id="manual", + data_schema=self.add_suggested_values_to_schema( + vol.Schema({vol.Required(CONF_DEVICE): str}), + user_input or {}, + ), + errors=errors, + ) + + +def get_ports() -> dict[str, str]: + """Get available serial ports keyed by their device path.""" + return { + port.device: human_readable_device_name( + os.path.realpath(port.device), + port.serial_number, + port.manufacturer, + port.description, + port.vid, + port.pid, + ) + for port in scan_serial_ports() + } diff --git a/homeassistant/components/denon_rs232/const.py b/homeassistant/components/denon_rs232/const.py new file mode 100644 index 00000000000..a408bd33509 --- /dev/null +++ b/homeassistant/components/denon_rs232/const.py @@ -0,0 +1,12 @@ +"""Constants for the Denon RS232 integration.""" + +import logging + +from denon_rs232 import DenonReceiver + +from homeassistant.config_entries import ConfigEntry + +LOGGER = logging.getLogger(__package__) +DOMAIN = "denon_rs232" + +type DenonRS232ConfigEntry = ConfigEntry[DenonReceiver] diff --git a/homeassistant/components/denon_rs232/manifest.json b/homeassistant/components/denon_rs232/manifest.json new file mode 100644 index 00000000000..eeca348feb2 --- /dev/null +++ b/homeassistant/components/denon_rs232/manifest.json @@ -0,0 +1,13 @@ +{ + "domain": "denon_rs232", + "name": "Denon RS232", + "codeowners": ["@balloob"], + "config_flow": true, + "dependencies": ["usb"], + "documentation": "https://www.home-assistant.io/integrations/denon_rs232", + "integration_type": "hub", + "iot_class": "local_push", + "loggers": ["denon_rs232"], + "quality_scale": "bronze", + "requirements": ["denon-rs232==3.0.0"] +} diff --git a/homeassistant/components/denon_rs232/media_player.py b/homeassistant/components/denon_rs232/media_player.py new file mode 100644 index 00000000000..ee40ce3cdaf --- /dev/null +++ b/homeassistant/components/denon_rs232/media_player.py @@ -0,0 +1,235 @@ +"""Media player platform for the Denon RS232 integration.""" + +from __future__ import annotations + +from typing import Literal, cast + +from denon_rs232 import ( + MIN_VOLUME_DB, + VOLUME_DB_RANGE, + DenonReceiver, + InputSource, + MainPlayer, + ReceiverState, + ZonePlayer, +) + +from homeassistant.components.media_player import ( + MediaPlayerDeviceClass, + MediaPlayerEntity, + MediaPlayerEntityFeature, + MediaPlayerState, +) +from homeassistant.core import HomeAssistant, callback +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from .const import DOMAIN, DenonRS232ConfigEntry + +INPUT_SOURCE_DENON_TO_HA: dict[InputSource, str] = { + InputSource.PHONO: "phono", + InputSource.CD: "cd", + InputSource.TUNER: "tuner", + InputSource.DVD: "dvd", + InputSource.VDP: "vdp", + InputSource.TV: "tv", + InputSource.DBS_SAT: "dbs_sat", + InputSource.VCR_1: "vcr_1", + InputSource.VCR_2: "vcr_2", + InputSource.VCR_3: "vcr_3", + InputSource.V_AUX: "v_aux", + InputSource.CDR_TAPE1: "cdr_tape1", + InputSource.MD_TAPE2: "md_tape2", + InputSource.HDP: "hdp", + InputSource.DVR: "dvr", + InputSource.TV_CBL: "tv_cbl", + InputSource.SAT: "sat", + InputSource.NET_USB: "net_usb", + InputSource.DOCK: "dock", + InputSource.IPOD: "ipod", + InputSource.BD: "bd", + InputSource.SAT_CBL: "sat_cbl", + InputSource.MPLAY: "mplay", + InputSource.GAME: "game", + InputSource.AUX1: "aux1", + InputSource.AUX2: "aux2", + InputSource.NET: "net", + InputSource.BT: "bt", + InputSource.USB_IPOD: "usb_ipod", + InputSource.EIGHT_K: "eight_k", + InputSource.PANDORA: "pandora", + InputSource.SIRIUSXM: "siriusxm", + InputSource.SPOTIFY: "spotify", + InputSource.FLICKR: "flickr", + InputSource.IRADIO: "iradio", + InputSource.SERVER: "server", + InputSource.FAVORITES: "favorites", + InputSource.LASTFM: "lastfm", + InputSource.XM: "xm", + InputSource.SIRIUS: "sirius", + InputSource.HDRADIO: "hdradio", + InputSource.DAB: "dab", +} + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: DenonRS232ConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up the Denon RS232 media player.""" + receiver = config_entry.runtime_data + entities = [DenonRS232MediaPlayer(receiver, receiver.main, config_entry, "main")] + + if receiver.zone_2.power is not None: + entities.append( + DenonRS232MediaPlayer(receiver, receiver.zone_2, config_entry, "zone_2") + ) + if receiver.zone_3.power is not None: + entities.append( + DenonRS232MediaPlayer(receiver, receiver.zone_3, config_entry, "zone_3") + ) + + async_add_entities(entities) + + +class DenonRS232MediaPlayer(MediaPlayerEntity): + """Representation of a Denon receiver controlled over RS232.""" + + _attr_device_class = MediaPlayerDeviceClass.RECEIVER + _attr_has_entity_name = True + _attr_translation_key = "receiver" + _attr_should_poll = False + + _volume_min = MIN_VOLUME_DB + _volume_range = VOLUME_DB_RANGE + + def __init__( + self, + receiver: DenonReceiver, + player: MainPlayer | ZonePlayer, + config_entry: DenonRS232ConfigEntry, + zone: Literal["main", "zone_2", "zone_3"], + ) -> None: + """Initialize the media player.""" + self._receiver = receiver + self._player = player + self._is_main = zone == "main" + + model = receiver.model + assert model is not None # We always set this + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, config_entry.entry_id)}, + manufacturer="Denon", + model=model.name, + name=config_entry.title, + ) + self._attr_unique_id = f"{config_entry.entry_id}_{zone}" + + self._attr_source_list = sorted( + INPUT_SOURCE_DENON_TO_HA[source] for source in model.input_sources + ) + self._attr_supported_features = ( + MediaPlayerEntityFeature.TURN_ON + | MediaPlayerEntityFeature.TURN_OFF + | MediaPlayerEntityFeature.VOLUME_SET + | MediaPlayerEntityFeature.VOLUME_STEP + | MediaPlayerEntityFeature.SELECT_SOURCE + ) + + if zone == "main": + self._attr_name = None + self._attr_supported_features |= MediaPlayerEntityFeature.VOLUME_MUTE + else: + self._attr_name = "Zone 2" if zone == "zone_2" else "Zone 3" + + self._async_update_from_player() + + async def async_added_to_hass(self) -> None: + """Subscribe to receiver state updates.""" + self.async_on_remove(self._receiver.subscribe(self._async_on_state_update)) + + @callback + def _async_on_state_update(self, state: ReceiverState | None) -> None: + """Handle a state update from the receiver.""" + if state is None: + self._attr_available = False + else: + self._attr_available = True + self._async_update_from_player() + self.async_write_ha_state() + + @callback + def _async_update_from_player(self) -> None: + """Update entity attributes from the shared player object.""" + if self._player.power is None: + self._attr_state = None + else: + self._attr_state = ( + MediaPlayerState.ON if self._player.power else MediaPlayerState.OFF + ) + + source = self._player.input_source + self._attr_source = INPUT_SOURCE_DENON_TO_HA.get(source) if source else None + + volume_min = self._player.volume_min + volume_max = self._player.volume_max + if volume_min is not None: + self._volume_min = volume_min + + if volume_max is not None and volume_max > volume_min: + self._volume_range = volume_max - volume_min + + volume = self._player.volume + if volume is not None: + self._attr_volume_level = (volume - self._volume_min) / self._volume_range + else: + self._attr_volume_level = None + + if self._is_main: + self._attr_is_volume_muted = cast(MainPlayer, self._player).mute + + async def async_turn_on(self) -> None: + """Turn the receiver on.""" + await self._player.power_on() + + async def async_turn_off(self) -> None: + """Turn the receiver off.""" + await self._player.power_standby() + + async def async_set_volume_level(self, volume: float) -> None: + """Set volume level, range 0..1.""" + db = volume * self._volume_range + self._volume_min + await self._player.set_volume(db) + + async def async_volume_up(self) -> None: + """Volume up.""" + await self._player.volume_up() + + async def async_volume_down(self) -> None: + """Volume down.""" + await self._player.volume_down() + + async def async_mute_volume(self, mute: bool) -> None: + """Mute or unmute.""" + player = cast(MainPlayer, self._player) + if mute: + await player.mute_on() + else: + await player.mute_off() + + async def async_select_source(self, source: str) -> None: + """Select input source.""" + input_source = next( + ( + input_source + for input_source, ha_source in INPUT_SOURCE_DENON_TO_HA.items() + if ha_source == source + ), + None, + ) + if input_source is None: + raise HomeAssistantError("Invalid source") + + await self._player.select_input_source(input_source) diff --git a/homeassistant/components/denon_rs232/quality_scale.yaml b/homeassistant/components/denon_rs232/quality_scale.yaml new file mode 100644 index 00000000000..b8c61a47e88 --- /dev/null +++ b/homeassistant/components/denon_rs232/quality_scale.yaml @@ -0,0 +1,60 @@ +rules: + # Bronze + action-setup: done + appropriate-polling: done + brands: done + common-modules: done + config-flow-test-coverage: done + config-flow: done + dependency-transparency: done + docs-actions: done + docs-high-level-description: done + docs-installation-instructions: done + docs-removal-instructions: done + entity-event-setup: done + entity-unique-id: done + has-entity-name: done + runtime-data: done + test-before-configure: done + test-before-setup: done + unique-config-entry: done + + # Silver + action-exceptions: todo + config-entry-unloading: todo + docs-configuration-parameters: todo + docs-installation-parameters: todo + entity-unavailable: todo + integration-owner: todo + log-when-unavailable: todo + parallel-updates: todo + reauthentication-flow: todo + test-coverage: todo + + # Gold + devices: todo + diagnostics: todo + discovery-update-info: todo + discovery: todo + docs-data-update: todo + docs-examples: todo + docs-known-limitations: todo + docs-supported-devices: todo + docs-supported-functions: todo + docs-troubleshooting: todo + docs-use-cases: todo + dynamic-devices: todo + entity-category: todo + entity-device-class: todo + entity-disabled-by-default: todo + entity-translations: todo + exception-translations: todo + icon-translations: todo + reconfiguration-flow: todo + repair-issues: todo + stale-devices: todo + + # Platinum + async-dependency: done + inject-websession: todo + strict-typing: todo diff --git a/homeassistant/components/denon_rs232/strings.json b/homeassistant/components/denon_rs232/strings.json new file mode 100644 index 00000000000..03db0a0d433 --- /dev/null +++ b/homeassistant/components/denon_rs232/strings.json @@ -0,0 +1,86 @@ +{ + "config": { + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]" + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "unknown": "[%key:common::config_flow::error::unknown%]" + }, + "step": { + "manual": { + "data": { + "device": "[%key:common::config_flow::data::port%]" + }, + "data_description": { + "device": "[%key:component::denon_rs232::config::step::user::data_description::device%]" + } + }, + "user": { + "data": { + "device": "[%key:common::config_flow::data::port%]", + "model": "Receiver model" + }, + "data_description": { + "device": "Serial port path to connect to", + "model": "Determines available features" + } + } + } + }, + "entity": { + "media_player": { + "receiver": { + "state_attributes": { + "source": { + "name": "Source", + "state": { + "aux1": "Aux 1", + "aux2": "Aux 2", + "bd": "BD Player", + "bt": "Bluetooth", + "cd": "CD", + "cdr_tape1": "CDR/Tape 1", + "dab": "DAB", + "dbs_sat": "DBS/Sat", + "dock": "Dock", + "dvd": "DVD", + "dvr": "DVR", + "eight_k": "8K", + "favorites": "Favorites", + "flickr": "Flickr", + "game": "Game", + "hdp": "HDP", + "hdradio": "HD Radio", + "ipod": "iPod", + "iradio": "Internet Radio", + "lastfm": "Last.fm", + "md_tape2": "MD/Tape 2", + "mplay": "Media Player", + "net": "HEOS Music", + "net_usb": "Network/USB", + "pandora": "Pandora", + "phono": "Phono", + "sat": "Sat", + "sat_cbl": "Satellite/Cable", + "server": "Server", + "sirius": "Sirius", + "siriusxm": "SiriusXM", + "spotify": "Spotify", + "tuner": "Tuner", + "tv": "TV Audio", + "tv_cbl": "TV/Cable", + "usb_ipod": "USB/iPod", + "v_aux": "V. Aux", + "vcr_1": "VCR 1", + "vcr_2": "VCR 2", + "vcr_3": "VCR 3", + "vdp": "VDP", + "xm": "XM" + } + } + } + } + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 9d223490e6b..1e91bdf9b40 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -142,6 +142,7 @@ FLOWS = { "deconz", "decora_wifi", "deluge", + "denon_rs232", "denonavr", "devialet", "devolo_home_control", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index 8f2a48cfdee..99bfc439851 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -1323,6 +1323,12 @@ "iot_class": "local_push", "name": "Denon AVR Network Receivers" }, + "denon_rs232": { + "integration_type": "hub", + "config_flow": true, + "iot_class": "local_push", + "name": "Denon RS232" + }, "heos": { "integration_type": "hub", "config_flow": true, diff --git a/requirements_all.txt b/requirements_all.txt index 1e980dce827..a5f6402f7d9 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -803,6 +803,9 @@ deluge-client==1.10.2 # homeassistant.components.lametric demetriek==1.3.0 +# homeassistant.components.denon_rs232 +denon-rs232==3.0.0 + # homeassistant.components.denonavr denonavr==1.3.2 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index bf9fb7b6a9e..8a080ee09de 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -715,6 +715,9 @@ deluge-client==1.10.2 # homeassistant.components.lametric demetriek==1.3.0 +# homeassistant.components.denon_rs232 +denon-rs232==3.0.0 + # homeassistant.components.denonavr denonavr==1.3.2 diff --git a/tests/components/denon_rs232/__init__.py b/tests/components/denon_rs232/__init__.py new file mode 100644 index 00000000000..d38597c2bee --- /dev/null +++ b/tests/components/denon_rs232/__init__.py @@ -0,0 +1,4 @@ +"""Tests for the Denon RS232 integration.""" + +MOCK_DEVICE = "/dev/ttyUSB0" +MOCK_MODEL = "avr_3805" diff --git a/tests/components/denon_rs232/conftest.py b/tests/components/denon_rs232/conftest.py new file mode 100644 index 00000000000..64dd2de9007 --- /dev/null +++ b/tests/components/denon_rs232/conftest.py @@ -0,0 +1,194 @@ +"""Test fixtures for the Denon RS232 integration.""" + +from __future__ import annotations + +from typing import Literal +from unittest.mock import AsyncMock, patch + +from denon_rs232 import ( + DenonReceiver, + DigitalInputMode, + InputSource, + ReceiverState, + TunerBand, + TunerMode, + ZoneState, +) +from denon_rs232.models import MODELS +import pytest + +from homeassistant.components.denon_rs232.const import DOMAIN +from homeassistant.const import CONF_DEVICE, CONF_MODEL +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + +from . import MOCK_DEVICE, MOCK_MODEL + +from tests.common import MockConfigEntry + +ZoneName = Literal["main", "zone_2", "zone_3"] + + +class MockMainStateView: + """Main-zone view over the receiver state.""" + + def __init__(self, state: MockState) -> None: + """Initialize the main-zone state view.""" + self._state = state + + @property + def power(self) -> bool | None: + """Return the main-zone power state.""" + return self._state.main_zone_power + + @power.setter + def power(self, value: bool | None) -> None: + self._state.main_zone_power = value + + @property + def input_source(self) -> InputSource | None: + """Return the main-zone input source.""" + return self._state.input_source + + @input_source.setter + def input_source(self, value: InputSource | None) -> None: + self._state.input_source = value + + @property + def volume(self) -> float | None: + """Return the main-zone volume.""" + return self._state.volume + + @volume.setter + def volume(self, value: float | None) -> None: + self._state.volume = value + + +class MockState(ReceiverState): + """Receiver state with helpers for zone-oriented tests.""" + + def get_zone(self, zone: ZoneName) -> MockMainStateView | ZoneState: + """Return the requested zone state view.""" + if zone == "main": + return MockMainStateView(self) + return getattr(self, zone) + + +class MockReceiver(DenonReceiver): + """Receiver test double built on the real receiver/player objects.""" + + def __init__(self, state: MockState) -> None: + """Initialize the mock receiver.""" + super().__init__(MOCK_DEVICE, model=MODELS[MOCK_MODEL]) + self._connected = True + self._load_state(state) + self._send_command = AsyncMock() + self._query = AsyncMock() + self.connect = AsyncMock(side_effect=self._mock_connect) + self.query_state = AsyncMock() + self.disconnect = AsyncMock(side_effect=self._mock_disconnect) + + def get_zone(self, zone: ZoneName): + """Return the matching live player object.""" + if zone == "main": + return self.main + if zone == "zone_2": + return self.zone_2 + return self.zone_3 + + def mock_state(self, state: MockState | None) -> None: + """Push a state update through the receiver.""" + self._connected = state is not None + if state is not None: + self._load_state(state) + self._notify_subscribers() + + async def _mock_connect(self) -> None: + """Pretend to open the serial connection.""" + self._connected = True + + async def _mock_disconnect(self) -> None: + """Pretend to close the serial connection.""" + self._connected = False + + def _load_state(self, state: MockState) -> None: + """Swap in a new state object and rebind the live players to it.""" + self._state = state + self.main._state = state + self.main._main_state = state + self.zone_2._state = state.zone_2 + self.zone_3._state = state.zone_3 + + +def _default_state() -> MockState: + """Return a ReceiverState with typical defaults.""" + return MockState( + power=True, + main_zone_power=True, + volume=-30.0, + volume_min=-80, + volume_max=10, + mute=False, + input_source=InputSource.CD, + surround_mode="STEREO", + digital_input=DigitalInputMode.AUTO, + tuner_band=TunerBand.FM, + tuner_mode=TunerMode.AUTO, + zone_2=ZoneState( + power=True, + input_source=InputSource.TUNER, + volume=-20.0, + ), + zone_3=ZoneState( + power=False, + input_source=InputSource.CD, + volume=-35.0, + ), + ) + + +@pytest.fixture +def initial_receiver_state(request: pytest.FixtureRequest) -> MockState: + """Return the initial receiver state for a test.""" + state = _default_state() + + if getattr(request, "param", None) == "main_only": + state.zone_2 = ZoneState() + state.zone_3 = ZoneState() + + return state + + +@pytest.fixture +def mock_receiver(initial_receiver_state: MockState) -> MockReceiver: + """Create a mock DenonReceiver.""" + return MockReceiver(initial_receiver_state) + + +@pytest.fixture +def mock_config_entry() -> MockConfigEntry: + """Create a mock config entry.""" + return MockConfigEntry( + domain=DOMAIN, + data={CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}, + title=MODELS[MOCK_MODEL].name, + ) + + +@pytest.fixture +async def init_components( + hass: HomeAssistant, + mock_receiver: MockReceiver, + mock_config_entry: MockConfigEntry, +) -> None: + """Initialize the Denon component.""" + hass.config.components.add("usb") + mock_config_entry.add_to_hass(hass) + with patch( + "homeassistant.components.denon_rs232.DenonReceiver", + return_value=mock_receiver, + ): + assert await async_setup_component(hass, DOMAIN, {}) + await hass.async_block_till_done() + await hass.async_block_till_done() + await hass.async_block_till_done() diff --git a/tests/components/denon_rs232/test_config_flow.py b/tests/components/denon_rs232/test_config_flow.py new file mode 100644 index 00000000000..1dc3d65f6c7 --- /dev/null +++ b/tests/components/denon_rs232/test_config_flow.py @@ -0,0 +1,244 @@ +"""Tests for the Denon RS232 config flow.""" + +from collections.abc import Generator +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from homeassistant.components.denon_rs232.config_flow import OPTION_PICK_MANUAL +from homeassistant.components.denon_rs232.const import DOMAIN +from homeassistant.components.usb import USBDevice +from homeassistant.config_entries import SOURCE_USER +from homeassistant.const import CONF_DEVICE, CONF_MODEL +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + +from . import MOCK_DEVICE, MOCK_MODEL + +from tests.common import MockConfigEntry, get_schema_suggested_value + + +@pytest.fixture(autouse=True) +def mock_list_serial_ports() -> Generator[list[USBDevice]]: + """Mock discovered serial ports.""" + ports = [ + USBDevice( + device=MOCK_DEVICE, + vid="123", + pid="456", + serial_number="mock-serial", + manufacturer="mock-manuf", + description=None, + ) + ] + + with patch( + "homeassistant.components.denon_rs232.config_flow.scan_serial_ports", + return_value=ports, + ): + yield ports + + +@pytest.fixture +def mock_async_setup_entry(mock_receiver: MagicMock) -> Generator[AsyncMock]: + """Prevent config-entry creation tests from setting up the integration.""" + + async def _mock_setup_entry(hass: HomeAssistant, entry) -> bool: + entry.runtime_data = mock_receiver + return True + + with patch( + "homeassistant.components.denon_rs232.async_setup_entry", + side_effect=_mock_setup_entry, + ) as mock_setup_entry: + yield mock_setup_entry + + +async def test_user_form_creates_entry( + hass: HomeAssistant, + mock_receiver: MagicMock, + mock_async_setup_entry: AsyncMock, +) -> None: + """Test successful config flow creates an entry.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + with patch( + "homeassistant.components.denon_rs232.config_flow.DenonReceiver", + return_value=mock_receiver, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == "AVR-3805 / AVC-3890" + assert result["data"] == {CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL} + mock_async_setup_entry.assert_awaited_once() + mock_receiver.connect.assert_awaited_once() + mock_receiver.disconnect.assert_awaited_once() + + +@pytest.mark.parametrize( + ("exception", "error"), + [ + (ConnectionError("No response"), "cannot_connect"), + (OSError("No such device"), "cannot_connect"), + (RuntimeError("boom"), "unknown"), + ], +) +async def test_user_form_error( + hass: HomeAssistant, + exception: Exception, + error: str, + mock_receiver: MagicMock, +) -> None: + """Test the user step reports connection and unexpected errors.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + mock_receiver.connect.side_effect = exception + + with patch( + "homeassistant.components.denon_rs232.config_flow.DenonReceiver", + return_value=mock_receiver, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + assert result["errors"] == {"base": error} + + +async def test_user_duplicate_port_aborts(hass: HomeAssistant) -> None: + """Test we abort if the same port is already configured.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}, + ) + entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_manual_form_creates_entry( + hass: HomeAssistant, + mock_receiver: MagicMock, + mock_async_setup_entry: AsyncMock, +) -> None: + """Test creating entry with manual user input.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE: OPTION_PICK_MANUAL, CONF_MODEL: MOCK_MODEL}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "manual" + + with patch( + "homeassistant.components.denon_rs232.config_flow.DenonReceiver", + return_value=mock_receiver, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE: MOCK_DEVICE}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == "AVR-3805 / AVC-3890" + assert result["data"] == {CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL} + mock_async_setup_entry.assert_awaited_once() + mock_receiver.connect.assert_awaited_once() + mock_receiver.disconnect.assert_awaited_once() + + +@pytest.mark.parametrize( + ("exception", "error"), + [ + (ValueError("Invalid port"), "cannot_connect"), + (ConnectionError("No response"), "cannot_connect"), + (OSError("No such device"), "cannot_connect"), + (RuntimeError("boom"), "unknown"), + ], +) +async def test_manual_form_error_handling( + hass: HomeAssistant, + exception: Exception, + error: str, + mock_receiver: MagicMock, +) -> None: + """Test the manual step reports connection and unexpected errors.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE: OPTION_PICK_MANUAL, CONF_MODEL: MOCK_MODEL}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "manual" + mock_receiver.connect.side_effect = exception + + with patch( + "homeassistant.components.denon_rs232.config_flow.DenonReceiver", + return_value=mock_receiver, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE: MOCK_DEVICE}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "manual" + assert result["errors"] == {"base": error} + assert ( + get_schema_suggested_value(result["data_schema"].schema, CONF_DEVICE) + == MOCK_DEVICE + ) + + +async def test_manual_duplicate_port_aborts(hass: HomeAssistant) -> None: + """Test we abort if the same port is already configured.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}, + ) + entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE: OPTION_PICK_MANUAL, CONF_MODEL: MOCK_MODEL}, + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_DEVICE: MOCK_DEVICE}, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" diff --git a/tests/components/denon_rs232/test_media_player.py b/tests/components/denon_rs232/test_media_player.py new file mode 100644 index 00000000000..8510ecf22f3 --- /dev/null +++ b/tests/components/denon_rs232/test_media_player.py @@ -0,0 +1,318 @@ +"""Tests for the Denon RS232 media player platform.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Literal +from unittest.mock import call + +from denon_rs232 import InputSource +import pytest + +from homeassistant.components.denon_rs232.media_player import INPUT_SOURCE_DENON_TO_HA +from homeassistant.components.media_player import ( + ATTR_INPUT_SOURCE, + ATTR_INPUT_SOURCE_LIST, + ATTR_MEDIA_VOLUME_LEVEL, + ATTR_MEDIA_VOLUME_MUTED, + DOMAIN as MP_DOMAIN, + SERVICE_SELECT_SOURCE, +) +from homeassistant.const import ( + ATTR_ENTITY_ID, + SERVICE_TURN_OFF, + SERVICE_TURN_ON, + SERVICE_VOLUME_DOWN, + SERVICE_VOLUME_MUTE, + SERVICE_VOLUME_SET, + SERVICE_VOLUME_UP, + STATE_OFF, + STATE_ON, + STATE_UNAVAILABLE, +) +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError + +from .conftest import MockReceiver, MockState, _default_state + +ZoneName = Literal["main", "zone_2", "zone_3"] + +MAIN_ENTITY_ID = "media_player.avr_3805_avc_3890" +ZONE_2_ENTITY_ID = "media_player.avr_3805_avc_3890_zone_2" +ZONE_3_ENTITY_ID = "media_player.avr_3805_avc_3890_zone_3" + +STRINGS_PATH = Path("homeassistant/components/denon_rs232/strings.json") + + +@pytest.fixture(autouse=True) +async def auto_init_components(init_components) -> None: + """Set up the component.""" + + +async def test_entities_created( + hass: HomeAssistant, mock_receiver: MockReceiver +) -> None: + """Test media player entities are created through config entry setup.""" + assert hass.states.get(MAIN_ENTITY_ID) is not None + assert hass.states.get(ZONE_2_ENTITY_ID) is not None + assert hass.states.get(ZONE_3_ENTITY_ID) is not None + mock_receiver.query_state.assert_awaited_once() + + +@pytest.mark.parametrize("initial_receiver_state", ["main_only"], indirect=True) +async def test_only_active_zones_are_created( + hass: HomeAssistant, initial_receiver_state: MockState +) -> None: + """Test setup only creates entities for zones with queried power state.""" + assert hass.states.get(MAIN_ENTITY_ID) is not None + assert hass.states.get(ZONE_2_ENTITY_ID) is None + assert hass.states.get(ZONE_3_ENTITY_ID) is None + + +@pytest.mark.parametrize( + ("zone", "entity_id", "initial_entity_state"), + [ + ("main", MAIN_ENTITY_ID, STATE_ON), + ("zone_2", ZONE_2_ENTITY_ID, STATE_ON), + ("zone_3", ZONE_3_ENTITY_ID, STATE_OFF), + ], +) +async def test_zone_state_updates( + hass: HomeAssistant, + mock_receiver: MockReceiver, + zone: ZoneName, + entity_id: str, + initial_entity_state: str, +) -> None: + """Test each zone updates from receiver pushes and disconnects.""" + assert hass.states.get(entity_id).state == initial_entity_state + + state = _default_state() + state.get_zone(zone).power = initial_entity_state != STATE_ON + mock_receiver.mock_state(state) + await hass.async_block_till_done() + + assert hass.states.get(entity_id).state != initial_entity_state + + mock_receiver.mock_state(None) + await hass.async_block_till_done() + + assert hass.states.get(entity_id).state == STATE_UNAVAILABLE + + +@pytest.mark.parametrize( + ("zone", "entity_id", "power_on_command", "power_off_command"), + [ + ("main", MAIN_ENTITY_ID, ("ZM", "ON"), ("ZM", "OFF")), + ("zone_2", ZONE_2_ENTITY_ID, ("Z2", "ON"), ("Z2", "OFF")), + ("zone_3", ZONE_3_ENTITY_ID, ("Z1", "ON"), ("Z1", "OFF")), + ], +) +async def test_power_controls( + hass: HomeAssistant, + mock_receiver: MockReceiver, + zone: ZoneName, + entity_id: str, + power_on_command: tuple[str, str], + power_off_command: tuple[str, str], +) -> None: + """Test power services send the right commands for each zone.""" + + await hass.services.async_call( + MP_DOMAIN, + SERVICE_TURN_ON, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + ) + assert mock_receiver._send_command.await_args == call(*power_on_command) + + await hass.services.async_call( + MP_DOMAIN, + SERVICE_TURN_OFF, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + ) + assert mock_receiver._send_command.await_args == call(*power_off_command) + + +@pytest.mark.parametrize( + ( + "zone", + "entity_id", + "initial_volume_level", + "set_command", + "volume_up_command", + "volume_down_command", + ), + [ + ( + "main", + MAIN_ENTITY_ID, + 50.0 / 90.0, + ("MV", "45"), + ("MV", "UP"), + ("MV", "DOWN"), + ), + ( + "zone_2", + ZONE_2_ENTITY_ID, + 60.0 / 90.0, + ("Z2", "45"), + ("Z2", "UP"), + ("Z2", "DOWN"), + ), + ], +) +async def test_volume_controls( + hass: HomeAssistant, + mock_receiver: MockReceiver, + zone: ZoneName, + entity_id: str, + initial_volume_level: float, + set_command: tuple[str, str], + volume_up_command: tuple[str, str], + volume_down_command: tuple[str, str], +) -> None: + """Test volume state and controls for each zone.""" + state = hass.states.get(entity_id) + + assert abs(state.attributes[ATTR_MEDIA_VOLUME_LEVEL] - initial_volume_level) < 0.001 + + await hass.services.async_call( + MP_DOMAIN, + SERVICE_VOLUME_SET, + {ATTR_ENTITY_ID: entity_id, ATTR_MEDIA_VOLUME_LEVEL: 0.5}, + blocking=True, + ) + assert mock_receiver._send_command.await_args == call(*set_command) + + await hass.services.async_call( + MP_DOMAIN, + SERVICE_VOLUME_UP, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + ) + assert mock_receiver._send_command.await_args == call(*volume_up_command) + + await hass.services.async_call( + MP_DOMAIN, + SERVICE_VOLUME_DOWN, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + ) + assert mock_receiver._send_command.await_args == call(*volume_down_command) + + +async def test_main_mute_controls( + hass: HomeAssistant, mock_receiver: MockReceiver +) -> None: + """Test mute state and controls for the main zone.""" + state = hass.states.get(MAIN_ENTITY_ID) + + assert state.attributes[ATTR_MEDIA_VOLUME_MUTED] is False + + await hass.services.async_call( + MP_DOMAIN, + SERVICE_VOLUME_MUTE, + {ATTR_ENTITY_ID: MAIN_ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: True}, + blocking=True, + ) + assert mock_receiver._send_command.await_args == call("MU", "ON") + + await hass.services.async_call( + MP_DOMAIN, + SERVICE_VOLUME_MUTE, + {ATTR_ENTITY_ID: MAIN_ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: False}, + blocking=True, + ) + assert mock_receiver._send_command.await_args == call("MU", "OFF") + + +@pytest.mark.parametrize( + ( + "zone", + "entity_id", + "initial_source", + "updated_source", + "expected_source", + "select_source_command", + ), + [ + ("main", MAIN_ENTITY_ID, "cd", InputSource.NET, "net", ("SI", "NET")), + ( + "zone_2", + ZONE_2_ENTITY_ID, + "tuner", + InputSource.BT, + "bt", + ("Z2", "BT"), + ), + ("zone_3", ZONE_3_ENTITY_ID, None, InputSource.DVD, "dvd", ("Z1", "DVD")), + ], +) +async def test_source_state_and_controls( + hass: HomeAssistant, + mock_receiver: MockReceiver, + zone: ZoneName, + entity_id: str, + initial_source: str | None, + updated_source: InputSource, + expected_source: str, + select_source_command: tuple[str, str], +) -> None: + """Test source state and selection for each zone.""" + entity_state = hass.states.get(entity_id) + + assert entity_state.attributes.get(ATTR_INPUT_SOURCE) == initial_source + + source_list = entity_state.attributes[ATTR_INPUT_SOURCE_LIST] + assert "cd" in source_list + assert "dvd" in source_list + assert "tuner" in source_list + assert source_list == sorted(source_list) + + state = _default_state() + zone_state = state.get_zone(zone) + zone_state.power = True + zone_state.input_source = updated_source + mock_receiver.mock_state(state) + await hass.async_block_till_done() + + assert hass.states.get(entity_id).attributes[ATTR_INPUT_SOURCE] == expected_source + + await hass.services.async_call( + MP_DOMAIN, + SERVICE_SELECT_SOURCE, + {ATTR_ENTITY_ID: entity_id, ATTR_INPUT_SOURCE: expected_source}, + blocking=True, + ) + assert mock_receiver._send_command.await_args == call(*select_source_command) + + +async def test_main_invalid_source_raises( + hass: HomeAssistant, +) -> None: + """Test invalid main-zone sources raise an error.""" + with pytest.raises(HomeAssistantError): + await hass.services.async_call( + MP_DOMAIN, + SERVICE_SELECT_SOURCE, + { + ATTR_ENTITY_ID: MAIN_ENTITY_ID, + ATTR_INPUT_SOURCE: "NONEXISTENT", + }, + blocking=True, + ) + + +def test_input_source_translation_keys_cover_all_enum_members() -> None: + """Test all input sources have a declared translation key.""" + assert set(INPUT_SOURCE_DENON_TO_HA) == set(InputSource) + + strings = json.loads(STRINGS_PATH.read_text("utf-8")) + assert set(INPUT_SOURCE_DENON_TO_HA.values()) == set( + strings["entity"]["media_player"]["receiver"]["state_attributes"]["source"][ + "state" + ] + )