Add Denon rs232 integration

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,58 @@
+"""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)
+        if receiver.connected:
+            await receiver.disconnect()
+        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,141 @@
+"""Config flow for the Denon RS232 integration."""
+
+from __future__ import annotations
+
+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(
+            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"]
+}