media_player.py

Go to the documentation of this file.

"""Support for interfacing with Russound via RNET Protocol."""
 
from __future__ import annotations
 
import logging
import math
 
from russound import russound
import voluptuous as vol
 
from homeassistant.components.media_player import (
    PLATFORM_SCHEMA as MEDIA_PLAYER_PLATFORM_SCHEMA,
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    MediaPlayerState,
)
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
_LOGGER = logging.getLogger(__name__)
 
CONF_ZONES = "zones"
CONF_SOURCES = "sources"
 
 
ZONE_SCHEMA = vol.Schema({vol.Required(CONF_NAME): cv.string})
 
SOURCE_SCHEMA = vol.Schema({vol.Required(CONF_NAME): cv.string})
 
PLATFORM_SCHEMA = MEDIA_PLAYER_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_HOST): cv.string,
        vol.Required(CONF_NAME): cv.string,
        vol.Required(CONF_PORT): cv.port,
        vol.Required(CONF_ZONES): vol.Schema({cv.positive_int: ZONE_SCHEMA}),
        vol.Required(CONF_SOURCES): vol.All(cv.ensure_list, [SOURCE_SCHEMA]),
    }
)
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the Russound RNET platform."""
    host = config.get(CONF_HOST)
    port = config.get(CONF_PORT)
 
    if host is None or port is None:
        _LOGGER.error("Invalid config. Expected %s and %s", CONF_HOST, CONF_PORT)
        return
 
    russ = russound.Russound(host, port)
    russ.connect()
 
    sources = [source["name"] for source in config[CONF_SOURCES]]
 
    if russ.is_connected():
        for zone_id, extra in config[CONF_ZONES].items():
            add_entities(
                [RussoundRNETDevice(hass, russ, sources, zone_id, extra)], True
            )
    else:
        _LOGGER.error("Not connected to %s:%s", host, port)
 
 
class RussoundRNETDevice(MediaPlayerEntity):
    """Representation of a Russound RNET device."""
 
    _attr_supported_features = (
        MediaPlayerEntityFeature.VOLUME_MUTE
        | MediaPlayerEntityFeature.VOLUME_SET
        | MediaPlayerEntityFeature.TURN_ON
        | MediaPlayerEntityFeature.TURN_OFF
        | MediaPlayerEntityFeature.SELECT_SOURCE
    )
 
    def __init__(self, hass, russ, sources, zone_id, extra):
        """Initialise the Russound RNET device."""
        self._attr_name_attr_name = extra["name"]
        self._russ_russ = russ
        self._attr_source_list_attr_source_list = sources
        # Each controller has a maximum of 6 zones, every increment of 6 zones
        # maps to an additional controller for easier backward compatibility
        self._controller_id_controller_id = str(math.ceil(zone_id / 6))
        # Each zone resets to 1-6 per controller
        self._zone_id_zone_id = (zone_id - 1) % 6 + 1
 
    def update(self) -> None:
        """Retrieve latest state."""
        # Updated this function to make a single call to get_zone_info, so that
        # with a single call we can get On/Off, Volume and Source, reducing the
        # amount of traffic and speeding up the update process.
        try:
            ret = self._russ_russ.get_zone_info(self._controller_id_controller_id, self._zone_id_zone_id, 4)
        except BrokenPipeError:
            _LOGGER.error("Broken Pipe Error, trying to reconnect to Russound RNET")
            self._russ_russ.connect()
            ret = self._russ_russ.get_zone_info(self._controller_id_controller_id, self._zone_id_zone_id, 4)
 
        _LOGGER.debug("ret= %s", ret)
        if ret is not None:
            _LOGGER.debug(
                "Updating status for RNET zone %s on controller %s",
                self._zone_id_zone_id,
                self._controller_id_controller_id,
            )
            if ret[0] == 0:
                self._attr_state_attr_state = MediaPlayerState.OFF
            else:
                self._attr_state_attr_state = MediaPlayerState.ON
            self._attr_volume_level_attr_volume_level = ret[2] * 2 / 100.0
            # Returns 0 based index for source.
            index = ret[1]
            # Possibility exists that user has defined list of all sources.
            # If a source is set externally that is beyond the defined list then
            # an exception will be thrown.
            # In this case return and unknown source (None)
            if self.source_listsource_list and 0 <= index < len(self.source_listsource_list):
                self._attr_source_attr_source = self.source_listsource_list[index]
        else:
            _LOGGER.error("Could not update status for zone %s", self._zone_id_zone_id)
 
    def set_volume_level(self, volume: float) -> None:
        """Set volume level.  Volume has a range (0..1).
 
        Translate this to a range of (0..100) as expected
        by _russ.set_volume()
        """
        self._russ_russ.set_volume(self._controller_id_controller_id, self._zone_id_zone_id, volume * 100)
 
    def turn_on(self) -> None:
        """Turn the media player on."""
        self._russ_russ.set_power(self._controller_id_controller_id, self._zone_id_zone_id, "1")
 
    def turn_off(self) -> None:
        """Turn off media player."""
        self._russ_russ.set_power(self._controller_id_controller_id, self._zone_id_zone_id, "0")
 
    def mute_volume(self, mute: bool) -> None:
        """Send mute command."""
        self._russ_russ.toggle_mute(self._controller_id_controller_id, self._zone_id_zone_id)
 
    def select_source(self, source: str) -> None:
        """Set the input source."""
        if self.source_listsource_list and source in self.source_listsource_list:
            index = self.source_listsource_list.index(source)
            # 0 based value for source
            self._russ_russ.set_source(self._controller_id_controller_id, self._zone_id_zone_id, index)