denon-bluetooth/custom_components/denon_bt/media_player.py

"""Media player entity for Denon AVR Bluetooth."""
from __future__ import annotations

import logging
from typing import Any

from homeassistant.components.media_player import (
    MediaPlayerDeviceClass,
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    MediaPlayerState,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_call_later

from .const import (
    CMD_MUTE_OFF,
    CMD_MUTE_ON,
    CMD_POWER_OFF,
    CMD_POWER_ON,
    CMD_VOLUME_DOWN,
    CMD_VOLUME_UP,
    DOMAIN,
    RECONNECT_INTERVAL,
    VOLUME_PRACTICAL_MAX,
)
from .protocol import DenonBTClient, DenonState, cmd_select_input

_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up the Denon BT media player from a config entry."""
    mac: str = entry.data["mac"]
    name: str = entry.data.get("name", "Denon AVR")
    async_add_entities([DenonBTMediaPlayer(hass, entry, mac, name)])


class DenonBTMediaPlayer(MediaPlayerEntity):
    """Representation of a Denon AVR-S540BT over Bluetooth RFCOMM."""

    _attr_device_class = MediaPlayerDeviceClass.RECEIVER
    _attr_has_entity_name = True
    _attr_name = None
    _attr_supported_features = (
        MediaPlayerEntityFeature.VOLUME_STEP
        | MediaPlayerEntityFeature.VOLUME_MUTE
        | MediaPlayerEntityFeature.TURN_ON
        | MediaPlayerEntityFeature.TURN_OFF
        | MediaPlayerEntityFeature.SELECT_SOURCE
    )

    def __init__(
        self,
        hass: HomeAssistant,
        entry: ConfigEntry,
        mac: str,
        name: str,
    ) -> None:
        self._hass = hass
        self._mac = mac

        self._attr_unique_id = mac.replace(":", "").lower()
        self._attr_device_info = {
            "identifiers": {(DOMAIN, mac)},
            "name": name,
            "manufacturer": "Denon",
            "model": "AVR-S540BT",
            "connections": {("bluetooth", mac)},
        }

        self._client: DenonBTClient | None = None
        self._state = DenonState()
        self._reconnect_cancel: callback | None = None
        self._reconnect_attempts = 0

    # ── Lifecycle ────────────────────────────────────────────────────────────

    async def async_added_to_hass(self) -> None:
        self._client = DenonBTClient(
            self._hass, self._mac, self._handle_state_update
        )
        await self._async_try_connect()

    async def async_will_remove_from_hass(self) -> None:
        self._cancel_reconnect()
        if self._client:
            await self._client.async_disconnect()

    # ── State callback from protocol client ──────────────────────────────────

    @callback
    def _handle_state_update(self, state: DenonState) -> None:
        """Called on the event loop when the protocol client reports a change."""
        self._state = state
        self._attr_available = self._client.connected if self._client else False

        if not self._attr_available:
            self._schedule_reconnect()
        else:
            self._reconnect_attempts = 0
            self._cancel_reconnect()

        self.async_write_ha_state()

    # ── Connection / reconnect ───────────────────────────────────────────────

    async def _async_try_connect(self) -> None:
        if self._client and await self._client.async_connect():
            self._attr_available = True
            self._reconnect_attempts = 0
            _LOGGER.info("Connected to Denon AVR at %s", self._mac)
        else:
            self._attr_available = False
            self._schedule_reconnect()
            _LOGGER.warning(
                "Could not connect to Denon AVR at %s", self._mac
            )

    def _schedule_reconnect(self) -> None:
        if self._reconnect_cancel is not None:
            return
        delay = min(RECONNECT_INTERVAL, 5 * (self._reconnect_attempts + 1))
        self._reconnect_cancel = async_call_later(
            self._hass, delay, self._async_reconnect
        )
        self._reconnect_attempts += 1
        _LOGGER.debug(
            "Reconnect to %s in %ds (attempt %d)",
            self._mac,
            delay,
            self._reconnect_attempts,
        )

    def _cancel_reconnect(self) -> None:
        if self._reconnect_cancel:
            self._reconnect_cancel()
            self._reconnect_cancel = None

    async def _async_reconnect(self, _now: Any = None) -> None:
        self._reconnect_cancel = None
        if self._client:
            await self._client.async_disconnect()
        await self._async_try_connect()

    # ── State properties ─────────────────────────────────────────────────────

    @property
    def state(self) -> MediaPlayerState:
        if not self._attr_available:
            return MediaPlayerState.OFF
        if self._state.power:
            return MediaPlayerState.ON
        return MediaPlayerState.OFF

    @property
    def volume_level(self) -> float | None:
        if self._state.volume_raw is None:
            return None
        return min(1.0, self._state.volume_raw / VOLUME_PRACTICAL_MAX)

    @property
    def is_volume_muted(self) -> bool | None:
        return self._state.muted

    @property
    def source(self) -> str | None:
        return self._state.source

    @property
    def source_list(self) -> list[str]:
        return list(self._state.sources.values())

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        attrs: dict[str, Any] = {}
        if self._state.sound_mode:
            attrs["sound_mode"] = self._state.sound_mode
        if self._state.volume_raw is not None:
            attrs["volume_raw"] = self._state.volume_raw
            attrs["volume_display"] = self._state.volume_raw / 2.0
        return attrs

    # ── Commands ─────────────────────────────────────────────────────────────

    async def async_turn_on(self) -> None:
        if self._client:
            if not self._client.connected:
                await self._async_try_connect()
            await self._client.async_send(CMD_POWER_ON)

    async def async_turn_off(self) -> None:
        if self._client:
            await self._client.async_send(CMD_POWER_OFF)

    async def async_volume_up(self) -> None:
        if self._client:
            await self._client.async_send(CMD_VOLUME_UP)

    async def async_volume_down(self) -> None:
        if self._client:
            await self._client.async_send(CMD_VOLUME_DOWN)

    async def async_mute_volume(self, mute: bool) -> None:
        if self._client:
            await self._client.async_send(CMD_MUTE_ON if mute else CMD_MUTE_OFF)

    async def async_select_source(self, source: str) -> None:
        if not self._client:
            return
        for code, name in self._state.sources.items():
            if name == source:
                await self._client.async_send(cmd_select_input(code))
                return
        _LOGGER.warning("Unknown source: %s", source)