Fixes

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/components/denon_rs232/__init__.py

@@ -0,0 +1,55 @@
+"""The Denon RS232 integration."""
+
+from __future__ import annotations
+
+from denon_rs232 import DenonReceiver, DenonState
+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()
+    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: DenonState | 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,124 @@
+"""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"
+
+
+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]})
+
+            model = MODELS[user_input[CONF_MODEL]]
+            receiver = DenonReceiver(user_input[CONF_DEVICE], model=model)
+            try:
+                await receiver.connect()
+            except ConnectionError, OSError:
+                errors["base"] = "cannot_connect"
+            except Exception:  # noqa: BLE001
+                LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                await receiver.disconnect()
+                return self.async_create_entry(title=model.name, data=user_input)
+
+        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]})
+
+            model = MODELS[self._model]
+            receiver = DenonReceiver(user_input[CONF_DEVICE], model=model)
+
+            try:
+                await receiver.connect()
+            except ConnectionError, OSError:
+                errors["base"] = "cannot_connect"
+            except Exception:  # noqa: BLE001
+                LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                await receiver.disconnect()
+                return self.async_create_entry(
+                    title=model.name,
+                    data={
+                        CONF_DEVICE: user_input[CONF_DEVICE],
+                        CONF_MODEL: self._model,
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="manual",
+            data_schema=vol.Schema({vol.Required(CONF_DEVICE): str}),
+            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==1.0.0"]
+}

homeassistant/components/denon_rs232/media_player.py

@@ -0,0 +1,202 @@
+"""Media player platform for the Denon RS232 integration."""
+
+from __future__ import annotations
+
+from denon_rs232 import (
+    MIN_VOLUME_DB,
+    VOLUME_DB_RANGE,
+    DenonReceiver,
+    DenonState,
+    InputSource,
+    PowerState,
+)
+
+from homeassistant.components.media_player import (
+    MediaPlayerDeviceClass,
+    MediaPlayerEntity,
+    MediaPlayerEntityFeature,
+    MediaPlayerState,
+)
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .const import DOMAIN, DenonRS232ConfigEntry
+
+POWER_STATE_DENON_TO_HA: dict[PowerState, MediaPlayerState] = {
+    PowerState.ON: MediaPlayerState.ON,
+    PowerState.STANDBY: MediaPlayerState.OFF,
+}
+
+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
+    async_add_entities([DenonRS232MediaPlayer(receiver, config_entry)])
+
+
+class DenonRS232MediaPlayer(MediaPlayerEntity):
+    """Representation of a Denon receiver controlled over RS232."""
+
+    _attr_device_class = MediaPlayerDeviceClass.RECEIVER
+    _attr_supported_features = (
+        MediaPlayerEntityFeature.TURN_ON
+        | MediaPlayerEntityFeature.TURN_OFF
+        | MediaPlayerEntityFeature.VOLUME_SET
+        | MediaPlayerEntityFeature.VOLUME_MUTE
+        | MediaPlayerEntityFeature.VOLUME_STEP
+        | MediaPlayerEntityFeature.SELECT_SOURCE
+    )
+    _attr_has_entity_name = True
+    _attr_name = None
+    _attr_translation_key = "receiver"
+    _attr_should_poll = False
+
+    def __init__(
+        self,
+        receiver: DenonReceiver,
+        config_entry: DenonRS232ConfigEntry,
+    ) -> None:
+        """Initialize the media player."""
+        self._receiver = receiver
+        self._attr_unique_id = config_entry.entry_id
+
+        model = receiver.model
+        model_name = model.name if model else None
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, config_entry.entry_id)},
+            manufacturer="Denon",
+            model=model_name,
+            name=config_entry.title,
+        )
+
+        if model:
+            self._attr_source_list = sorted(
+                INPUT_SOURCE_DENON_TO_HA[source] for source in model.input_sources
+            )
+        else:
+            self._attr_source_list = sorted(
+                INPUT_SOURCE_DENON_TO_HA[source] for source in InputSource
+            )
+
+        self._volume_min = MIN_VOLUME_DB
+        self._volume_range = VOLUME_DB_RANGE
+        self._async_update_from_state(receiver.state)
+
+    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: DenonState | 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_state(state)
+        self.async_write_ha_state()
+
+    @callback
+    def _async_update_from_state(self, state: DenonState) -> None:
+        """Update entity attributes from a DenonState snapshot."""
+        self._attr_state = POWER_STATE_DENON_TO_HA.get(state.power)
+        self._attr_source = INPUT_SOURCE_DENON_TO_HA.get(state.input_source)
+        self._attr_is_volume_muted = state.mute
+
+        if state.volume_min is not None:
+            self._volume_min = state.volume_min
+
+            if state.volume_max is not None and state.volume_max > state.volume_min:
+                self._volume_range = state.volume_max - state.volume_min
+
+        if state.volume is not None:
+            self._attr_volume_level = (
+                state.volume - self._volume_min
+            ) / self._volume_range
+        else:
+            self._attr_volume_level = None
+
+    async def async_turn_on(self) -> None:
+        """Turn the receiver on."""
+        await self._receiver.power_on()
+
+    async def async_turn_off(self) -> None:
+        """Turn the receiver off."""
+        await self._receiver.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._receiver.set_volume(db)
+
+    async def async_volume_up(self) -> None:
+        """Volume up."""
+        await self._receiver.volume_up()
+
+    async def async_volume_down(self) -> None:
+        """Volume down."""
+        await self._receiver.volume_down()
+
+    async def async_mute_volume(self, mute: bool) -> None:
+        """Mute or unmute."""
+        if mute:
+            await self._receiver.mute_on()
+        else:
+            await self._receiver.mute_off()
+
+    async def async_select_source(self, source: str) -> None:
+        """Select input source."""
+        for input_source, ha_source in INPUT_SOURCE_DENON_TO_HA.items():
+            if ha_source == source:
+                await self._receiver.select_input_source(input_source)
+                break

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,85 @@
+{
+  "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"
+        }
+      }
+    }
+  },
+  "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

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

homeassistant/generated/integrations.json

@@ -1325,6 +1325,12 @@
         }
       }
     },
+    "denon_rs232": {
+      "name": "Denon RS232",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "local_push"
+    },
     "devialet": {
       "name": "Devialet",
       "integration_type": "device",

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==1.0.0
+
 # homeassistant.components.denonavr
 denonavr==1.3.2
 

requirements_test_all.txt

@@ -712,6 +712,9 @@ deluge-client==1.10.2
 # homeassistant.components.lametric
 demetriek==1.3.0
 
+# homeassistant.components.denon_rs232
+denon-rs232==1.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,105 @@
+"""Test fixtures for the Denon RS232 integration."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from denon_rs232 import (
+    DenonReceiver,
+    DenonState,
+    DigitalInputMode,
+    InputSource,
+    PowerState,
+    TunerBand,
+    TunerMode,
+)
+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 tests.common import MockConfigEntry
+
+from . import MOCK_DEVICE, MOCK_MODEL
+
+
+def _default_state() -> DenonState:
+    """Return a DenonState with typical defaults."""
+    return DenonState(
+        power=PowerState.ON,
+        main_zone=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,
+    )
+
+
+@pytest.fixture
+def mock_receiver() -> MagicMock:
+    """Create a mock DenonReceiver."""
+    receiver = MagicMock(spec=DenonReceiver)
+    receiver.connect = AsyncMock()
+    receiver.disconnect = AsyncMock()
+    receiver.power_on = AsyncMock()
+    receiver.power_standby = AsyncMock()
+    receiver.set_volume = AsyncMock()
+    receiver.volume_up = AsyncMock()
+    receiver.volume_down = AsyncMock()
+    receiver.mute_on = AsyncMock()
+    receiver.mute_off = AsyncMock()
+    receiver.select_input_source = AsyncMock()
+    receiver.set_surround_mode = AsyncMock()
+    receiver.connected = True
+    receiver.state = _default_state()
+    receiver.model = MODELS[MOCK_MODEL]
+
+    subscribers = receiver._subscribers = []
+
+    def subscribe(callback):
+        subscribers.append(callback)
+        return lambda: subscribers.remove(callback)
+
+    receiver.subscribe = subscribe
+
+    def mock_state(state: DenonState | None) -> None:
+        receiver.state = state
+        for sub in list(subscribers):
+            sub(state)
+
+    receiver.mock_state = mock_state
+
+    return receiver
+
+
+@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: MagicMock, 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,231 @@
+"""Tests for the Denon RS232 config flow."""
+
+from collections.abc import Generator
+from unittest.mock import AsyncMock, 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 tests.common import MockConfigEntry
+
+from . import MOCK_DEVICE, MOCK_MODEL
+
+
+@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
+
+
+async def test_user_form(hass: HomeAssistant) -> None:
+    """Test we show the user form."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+
+async def test_user_form_creates_entry(hass: HomeAssistant) -> None:
+    """Test successful config flow creates an entry."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    mock_receiver = AsyncMock()
+    mock_receiver.connect = AsyncMock()
+    mock_receiver.disconnect = AsyncMock()
+
+    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_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
+) -> 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 = AsyncMock()
+    mock_receiver.connect = AsyncMock(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) -> 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"
+
+    mock_receiver = AsyncMock()
+    mock_receiver.connect = AsyncMock()
+    mock_receiver.disconnect = AsyncMock()
+
+    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_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_manual_form_error_handling(
+    hass: HomeAssistant, exception: Exception, error: str
+) -> 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 = AsyncMock()
+    mock_receiver.connect = AsyncMock(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}
+
+
+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,313 @@
+"""Tests for the Denon RS232 media player platform."""
+
+import json
+from pathlib import Path
+
+from denon_rs232 import InputSource, PowerState
+import pytest
+
+from homeassistant.components.denon_rs232.media_player import (
+    INPUT_SOURCE_DENON_TO_HA,
+    POWER_STATE_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,
+)
+from homeassistant.core import HomeAssistant
+
+from .conftest import _default_state
+
+ENTITY_ID = "media_player.avr_3805_avc_3890"
+STRINGS_PATH = Path("homeassistant/components/denon_rs232/strings.json")
+
+
+def _translation_state_keys(attribute: str) -> set[str]:
+    """Return the translation keys declared for a state attribute."""
+    strings = json.loads(STRINGS_PATH.read_text())
+    return set(
+        strings["entity"]["media_player"]["receiver"]["state_attributes"][attribute][
+            "state"
+        ]
+    )
+
+
+@pytest.fixture(autouse=True)
+async def auto_init_components(init_components) -> None:
+    """Set up the component."""
+
+
+async def test_entity_created(hass: HomeAssistant) -> None:
+    """Test media player entity is created with correct state."""
+    state = hass.states.get(ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_ON
+
+
+async def test_state_on(hass: HomeAssistant) -> None:
+    """Test state is ON when receiver is powered on."""
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == STATE_ON
+
+
+async def test_state_off(hass: HomeAssistant, mock_receiver) -> None:
+    """Test state is OFF when receiver is in standby."""
+    new_state = _default_state()
+    new_state.power = PowerState.STANDBY
+    mock_receiver.mock_state(new_state)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == STATE_OFF
+
+
+async def test_volume_level(hass: HomeAssistant) -> None:
+    """Test volume level is correctly converted from dB to 0..1."""
+    # -30 dB: ((-30) - (-80)) / 90 = 50 / 90 ≈ 0.5555
+    state = hass.states.get(ENTITY_ID)
+    volume = state.attributes[ATTR_MEDIA_VOLUME_LEVEL]
+    assert abs(volume - 50.0 / 90.0) < 0.001
+
+
+async def test_mute_state(hass: HomeAssistant) -> None:
+    """Test mute state is reported."""
+    state = hass.states.get(ENTITY_ID)
+    assert state.attributes[ATTR_MEDIA_VOLUME_MUTED] is False
+
+
+async def test_source(hass: HomeAssistant) -> None:
+    """Test current source is reported."""
+    state = hass.states.get(ENTITY_ID)
+    assert state.attributes[ATTR_INPUT_SOURCE] == "cd"
+
+
+async def test_source_net(hass: HomeAssistant, mock_receiver) -> None:
+    """Test NET source is reported with the translation key."""
+    new_state = _default_state()
+    new_state.input_source = InputSource.NET
+    mock_receiver.mock_state(new_state)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.attributes[ATTR_INPUT_SOURCE] == "net"
+
+
+async def test_source_bluetooth(hass: HomeAssistant, mock_receiver) -> None:
+    """Test BT source is reported with the translation key."""
+    new_state = _default_state()
+    new_state.input_source = InputSource.BT
+    mock_receiver.mock_state(new_state)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.attributes[ATTR_INPUT_SOURCE] == "bt"
+
+
+async def test_source_list(hass: HomeAssistant) -> None:
+    """Test source list comes from the model definition."""
+    state = hass.states.get(ENTITY_ID)
+    source_list = 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)
+
+
+async def test_sound_mode_not_exposed(hass: HomeAssistant) -> None:
+    """Test surround mode is not exposed in Home Assistant."""
+    state = hass.states.get(ENTITY_ID)
+    assert "sound_mode" not in state.attributes
+    assert "sound_mode_list" not in state.attributes
+
+
+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)
+    assert set(INPUT_SOURCE_DENON_TO_HA.values()) == _translation_state_keys("source")
+
+
+def test_power_state_mapping_covers_all_values() -> None:
+    """Test all power states have a media player state mapping."""
+    assert set(POWER_STATE_DENON_TO_HA) == set(PowerState)
+
+
+async def test_turn_on(hass: HomeAssistant, mock_receiver) -> None:
+    """Test turning on the receiver."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_TURN_ON,
+        {ATTR_ENTITY_ID: ENTITY_ID},
+        blocking=True,
+    )
+    mock_receiver.power_on.assert_awaited_once()
+
+
+async def test_turn_off(hass: HomeAssistant, mock_receiver) -> None:
+    """Test turning off the receiver."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_TURN_OFF,
+        {ATTR_ENTITY_ID: ENTITY_ID},
+        blocking=True,
+    )
+    mock_receiver.power_standby.assert_awaited_once()
+
+
+async def test_set_volume(hass: HomeAssistant, mock_receiver) -> None:
+    """Test setting volume level converts from 0..1 to dB."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_VOLUME_SET,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_LEVEL: 0.5},
+        blocking=True,
+    )
+    # 0.5 * 90 + (-80) = -35.0
+    mock_receiver.set_volume.assert_awaited_once_with(-35.0)
+
+
+async def test_volume_up(hass: HomeAssistant, mock_receiver) -> None:
+    """Test volume up."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_VOLUME_UP,
+        {ATTR_ENTITY_ID: ENTITY_ID},
+        blocking=True,
+    )
+    mock_receiver.volume_up.assert_awaited_once()
+
+
+async def test_volume_down(hass: HomeAssistant, mock_receiver) -> None:
+    """Test volume down."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_VOLUME_DOWN,
+        {ATTR_ENTITY_ID: ENTITY_ID},
+        blocking=True,
+    )
+    mock_receiver.volume_down.assert_awaited_once()
+
+
+async def test_mute(hass: HomeAssistant, mock_receiver) -> None:
+    """Test muting."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_VOLUME_MUTE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: True},
+        blocking=True,
+    )
+    mock_receiver.mute_on.assert_awaited_once()
+
+
+async def test_unmute(hass: HomeAssistant, mock_receiver) -> None:
+    """Test unmuting."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_VOLUME_MUTE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: False},
+        blocking=True,
+    )
+    mock_receiver.mute_off.assert_awaited_once()
+
+
+async def test_select_source(hass: HomeAssistant, mock_receiver) -> None:
+    """Test selecting input source."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "dvd"},
+        blocking=True,
+    )
+    mock_receiver.select_input_source.assert_awaited_once_with(InputSource.DVD)
+
+
+async def test_select_source_net(hass: HomeAssistant, mock_receiver) -> None:
+    """Test selecting NET source."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "net"},
+        blocking=True,
+    )
+    mock_receiver.select_input_source.assert_awaited_once_with(InputSource.NET)
+
+
+async def test_select_source_bluetooth(hass: HomeAssistant, mock_receiver) -> None:
+    """Test selecting BT source."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "bt"},
+        blocking=True,
+    )
+    mock_receiver.select_input_source.assert_awaited_once_with(InputSource.BT)
+
+
+async def test_select_source_raw_value_is_ignored(
+    hass: HomeAssistant, mock_receiver
+) -> None:
+    """Test selecting a raw protocol source value does nothing."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "DVD"},
+        blocking=True,
+    )
+    mock_receiver.select_input_source.assert_not_awaited()
+
+
+async def test_select_source_unknown(hass: HomeAssistant, mock_receiver) -> None:
+    """Test selecting an unknown source does nothing."""
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "NONEXISTENT"},
+        blocking=True,
+    )
+    mock_receiver.select_input_source.assert_not_awaited()
+
+
+async def test_push_update(hass: HomeAssistant, mock_receiver) -> None:
+    """Test state updates from the receiver via subscribe callback."""
+    new_state = _default_state()
+    new_state.volume = -20.0
+    new_state.input_source = InputSource.DVD
+
+    mock_receiver.mock_state(new_state)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.attributes[ATTR_INPUT_SOURCE] == "dvd"
+    assert "sound_mode" not in state.attributes
+    expected_volume = ((-20.0) - (-80.0)) / 90.0
+    assert abs(state.attributes[ATTR_MEDIA_VOLUME_LEVEL] - expected_volume) < 0.001
+
+
+async def test_disconnect(hass: HomeAssistant, mock_receiver) -> None:
+    """Test entity becomes unavailable after disconnect."""
+    mock_receiver.mock_state(None)
+    await hass.async_block_till_done()
+
+    assert hass.states.get(ENTITY_ID).state == "unavailable"
+
+
+async def test_unload(hass: HomeAssistant, mock_receiver, mock_config_entry) -> None:
+    """Test unloading the integration disconnects the receiver."""
+    await hass.config_entries.async_unload(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    mock_receiver.disconnect.assert_awaited_once()