add denon rs232 integration

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

homeassistant/brands/denon.json

@@ -1,5 +1,5 @@
 {
   "domain": "denon",
   "name": "Denon",
-  "integrations": ["denon", "denonavr", "heos"]
+  "integrations": ["denon", "denonavr", "denon_rs232", "heos"]
 }

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

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()
+    }

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]

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"]
+}

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)

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

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"
+            }
+          }
+        }
+      }
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -142,6 +142,7 @@ FLOWS = {
         "deconz",
         "decora_wifi",
         "deluge",
+        "denon_rs232",
         "denonavr",
         "devialet",
         "devolo_home_control",

homeassistant/generated/integrations.json

@@ -1317,6 +1317,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,

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
 

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
 

tests/components/denon_rs232/__init__.py

@@ -0,0 +1,4 @@
+"""Tests for the Denon RS232 integration."""
+
+MOCK_DEVICE = "/dev/ttyUSB0"
+MOCK_MODEL = "avr_3805"

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()

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"

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"
+        ]
+    )