hass-core/homeassistant/components/hdmi_cec.py

"""
HDMI CEC component.

For more details about this component, please refer to the documentation at
https://home-assistant.io/components/hdmi_cec/
"""
import logging
import multiprocessing
import os
from collections import defaultdict
from functools import reduce

import voluptuous as vol

import homeassistant.helpers.config_validation as cv
from homeassistant import core
from homeassistant.components import discovery
from homeassistant.components.media_player import DOMAIN as MEDIA_PLAYER
from homeassistant.components.switch import DOMAIN as SWITCH
from homeassistant.config import load_yaml_config_file
from homeassistant.const import (EVENT_HOMEASSISTANT_START, STATE_UNKNOWN,
                                 EVENT_HOMEASSISTANT_STOP, STATE_ON,
                                 STATE_OFF, CONF_DEVICES, CONF_PLATFORM,
                                 CONF_CUSTOMIZE, STATE_PLAYING, STATE_IDLE,
                                 STATE_PAUSED, CONF_HOST)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity import Entity

REQUIREMENTS = ['pyCEC==0.4.9']

DOMAIN = 'hdmi_cec'

_LOGGER = logging.getLogger(__name__)

DEFAULT_DISPLAY_NAME = "HomeAssistant"

ICON_UNKNOWN = 'mdi:help'
ICON_AUDIO = 'mdi:speaker'
ICON_PLAYER = 'mdi:play'
ICON_TUNER = 'mdi:nest-thermostat'
ICON_RECORDER = 'mdi:microphone'
ICON_TV = 'mdi:television'
ICONS_BY_TYPE = {
    0: ICON_TV,
    1: ICON_RECORDER,
    3: ICON_TUNER,
    4: ICON_PLAYER,
    5: ICON_AUDIO
}

CEC_DEVICES = defaultdict(list)

CMD_UP = 'up'
CMD_DOWN = 'down'
CMD_MUTE = 'mute'
CMD_UNMUTE = 'unmute'
CMD_MUTE_TOGGLE = 'toggle mute'
CMD_PRESS = 'press'
CMD_RELEASE = 'release'

EVENT_CEC_COMMAND_RECEIVED = 'cec_command_received'
EVENT_CEC_KEYPRESS_RECEIVED = 'cec_keypress_received'

ATTR_PHYSICAL_ADDRESS = 'physical_address'
ATTR_TYPE_ID = 'type_id'
ATTR_VENDOR_NAME = 'vendor_name'
ATTR_VENDOR_ID = 'vendor_id'
ATTR_DEVICE = 'device'
ATTR_COMMAND = 'command'
ATTR_TYPE = 'type'
ATTR_KEY = 'key'
ATTR_DUR = 'dur'
ATTR_SRC = 'src'
ATTR_DST = 'dst'
ATTR_CMD = 'cmd'
ATTR_ATT = 'att'
ATTR_RAW = 'raw'
ATTR_DIR = 'dir'
ATTR_ABT = 'abt'
ATTR_NEW = 'new'
ATTR_ON = 'on'
ATTR_OFF = 'off'
ATTR_TOGGLE = 'toggle'

_VOL_HEX = vol.Any(vol.Coerce(int), lambda x: int(x, 16))

SERVICE_SEND_COMMAND = 'send_command'
SERVICE_SEND_COMMAND_SCHEMA = vol.Schema({
    vol.Optional(ATTR_CMD): _VOL_HEX,
    vol.Optional(ATTR_SRC): _VOL_HEX,
    vol.Optional(ATTR_DST): _VOL_HEX,
    vol.Optional(ATTR_ATT): _VOL_HEX,
    vol.Optional(ATTR_RAW): vol.Coerce(str)
}, extra=vol.PREVENT_EXTRA)

SERVICE_VOLUME = 'volume'
SERVICE_VOLUME_SCHEMA = vol.Schema({
    vol.Optional(CMD_UP): vol.Any(CMD_PRESS, CMD_RELEASE, vol.Coerce(int)),
    vol.Optional(CMD_DOWN): vol.Any(CMD_PRESS, CMD_RELEASE, vol.Coerce(int)),
    vol.Optional(CMD_MUTE): vol.Any(ATTR_ON, ATTR_OFF, ATTR_TOGGLE),
}, extra=vol.PREVENT_EXTRA)

SERVICE_UPDATE_DEVICES = 'update'
SERVICE_UPDATE_DEVICES_SCHEMA = vol.Schema({
    DOMAIN: vol.Schema({})
}, extra=vol.PREVENT_EXTRA)

SERVICE_SELECT_DEVICE = 'select_device'

SERVICE_POWER_ON = 'power_on'
SERVICE_STANDBY = 'standby'

# pylint: disable=unnecessary-lambda
DEVICE_SCHEMA = vol.Schema({
    vol.All(cv.positive_int): vol.Any(lambda devices: DEVICE_SCHEMA(devices),
                                      cv.string)
})

CUSTOMIZE_SCHEMA = vol.Schema({
    vol.Optional(CONF_PLATFORM, default=MEDIA_PLAYER): vol.Any(MEDIA_PLAYER,
                                                               SWITCH)
})

CONF_DISPLAY_NAME = 'osd_name'
CONFIG_SCHEMA = vol.Schema({
    DOMAIN: vol.Schema({
        vol.Optional(CONF_DEVICES): vol.Any(DEVICE_SCHEMA,
                                            vol.Schema({
                                                vol.All(cv.string): vol.Any(
                                                    cv.string)
                                            })),
        vol.Optional(CONF_PLATFORM): vol.Any(SWITCH, MEDIA_PLAYER),
        vol.Optional(CONF_HOST): cv.string,
        vol.Optional(CONF_DISPLAY_NAME): cv.string,
    })
}, extra=vol.ALLOW_EXTRA)


def pad_physical_address(addr):
    """Right-pad a physical address."""
    return addr + [0] * (4 - len(addr))


def parse_mapping(mapping, parents=None):
    """Parse configuration device mapping."""
    if parents is None:
        parents = []
    for addr, val in mapping.items():
        if isinstance(addr, (str,)) and isinstance(val, (str,)):
            from pycec.network import PhysicalAddress
            yield (addr, PhysicalAddress(val))
        else:
            cur = parents + [addr]
            if isinstance(val, dict):
                yield from parse_mapping(val, cur)
            elif isinstance(val, str):
                yield (val, pad_physical_address(cur))


def setup(hass: HomeAssistant, base_config):
    """Setup CEC capability."""
    from pycec.network import HDMINetwork
    from pycec.commands import CecCommand, KeyReleaseCommand, KeyPressCommand
    from pycec.const import KEY_VOLUME_UP, KEY_VOLUME_DOWN, KEY_MUTE_ON, \
        KEY_MUTE_OFF, KEY_MUTE_TOGGLE, ADDR_AUDIOSYSTEM, ADDR_BROADCAST, \
        ADDR_UNREGISTERED
    from pycec.cec import CecAdapter
    from pycec.tcp import TcpAdapter

    # Parse configuration into a dict of device name to physical address
    # represented as a list of four elements.
    device_aliases = {}
    devices = base_config[DOMAIN].get(CONF_DEVICES, {})
    _LOGGER.debug("Parsing config %s", devices)
    device_aliases.update(parse_mapping(devices))
    _LOGGER.debug("Parsed devices: %s", device_aliases)

    platform = base_config[DOMAIN].get(CONF_PLATFORM, SWITCH)

    loop = (
        # Create own thread if more than 1 CPU
        hass.loop if multiprocessing.cpu_count() < 2 else None)
    host = base_config[DOMAIN].get(CONF_HOST, None)
    display_name = base_config[DOMAIN].get(CONF_DISPLAY_NAME,
                                           DEFAULT_DISPLAY_NAME)
    if host:
        adapter = TcpAdapter(host, name=display_name, activate_source=False)
    else:
        adapter = CecAdapter(name=display_name, activate_source=False)
    hdmi_network = HDMINetwork(adapter, loop=loop)

    def _volume(call):
        """Increase/decrease volume and mute/unmute system."""
        mute_key_mapping = {ATTR_TOGGLE: KEY_MUTE_TOGGLE, ATTR_ON: KEY_MUTE_ON,
                            ATTR_OFF: KEY_MUTE_OFF}
        for cmd, att in call.data.items():
            if cmd == CMD_UP:
                _process_volume(KEY_VOLUME_UP, att)
            elif cmd == CMD_DOWN:
                _process_volume(KEY_VOLUME_DOWN, att)
            elif cmd == CMD_MUTE:
                hdmi_network.send_command(
                    KeyPressCommand(mute_key_mapping[att],
                                    dst=ADDR_AUDIOSYSTEM))
                hdmi_network.send_command(
                    KeyReleaseCommand(dst=ADDR_AUDIOSYSTEM))
                _LOGGER.info("Audio muted")
            else:
                _LOGGER.warning("Unknown command %s", cmd)

    def _process_volume(cmd, att):
        if isinstance(att, (str,)):
            att = att.strip()
        if att == CMD_PRESS:
            hdmi_network.send_command(
                KeyPressCommand(cmd, dst=ADDR_AUDIOSYSTEM))
        elif att == CMD_RELEASE:
            hdmi_network.send_command(KeyReleaseCommand(dst=ADDR_AUDIOSYSTEM))
        else:
            att = 1 if att == "" else int(att)
            for _ in range(0, att):
                hdmi_network.send_command(
                    KeyPressCommand(cmd, dst=ADDR_AUDIOSYSTEM))
                hdmi_network.send_command(
                    KeyReleaseCommand(dst=ADDR_AUDIOSYSTEM))

    def _tx(call):
        """Send CEC command."""
        data = call.data
        if ATTR_RAW in data:
            command = CecCommand(data[ATTR_RAW])
        else:
            if ATTR_SRC in data:
                src = data[ATTR_SRC]
            else:
                src = ADDR_UNREGISTERED
            if ATTR_DST in data:
                dst = data[ATTR_DST]
            else:
                dst = ADDR_BROADCAST
            if ATTR_CMD in data:
                cmd = data[ATTR_CMD]
            else:
                _LOGGER.error("Attribute 'cmd' is missing")
                return False
            if ATTR_ATT in data:
                if isinstance(data[ATTR_ATT], (list,)):
                    att = data[ATTR_ATT]
                else:
                    att = reduce(lambda x, y: "%s:%x" % (x, y), data[ATTR_ATT])
            else:
                att = ""
            command = CecCommand(cmd, dst, src, att)
        hdmi_network.send_command(command)

    @callback
    def _standby(call):
        hdmi_network.standby()

    @callback
    def _power_on(call):
        hdmi_network.power_on()

    def _select_device(call):
        """Select the active device."""
        from pycec.network import PhysicalAddress

        addr = call.data[ATTR_DEVICE]
        if not addr:
            _LOGGER.error("Device not found: %s", call.data[ATTR_DEVICE])
            return
        if addr in device_aliases:
            addr = device_aliases[addr]
        else:
            entity = hass.states.get(addr)
            _LOGGER.debug("Selecting entity %s", entity)
            if entity is not None:
                addr = entity.attributes['physical_address']
                _LOGGER.debug("Address acquired: %s", addr)
                if addr is None:
                    _LOGGER.error("Device %s has not physical address.",
                                  call.data[ATTR_DEVICE])
                    return
        if not isinstance(addr, (PhysicalAddress,)):
            addr = PhysicalAddress(addr)
        hdmi_network.active_source(addr)
        _LOGGER.info("Selected %s (%s)", call.data[ATTR_DEVICE], addr)

    def _update(call):
        """
        Callback called when device update is needed.

        - called by service, requests CEC network to update data.
        """
        hdmi_network.scan()

    @callback
    def _new_device(device):
        """Called when new device is detected by HDMI network."""
        key = DOMAIN + '.' + device.name
        hass.data[key] = device
        discovery.load_platform(hass, base_config.get(core.DOMAIN).get(
            CONF_CUSTOMIZE, {}).get(key, {}).get(CONF_PLATFORM, platform),
                                DOMAIN, discovered={ATTR_NEW: [key]},
                                hass_config=base_config)

    def _shutdown(call):
        hdmi_network.stop()

    def _start_cec(event):
        """Register services and start HDMI network to watch for devices."""
        descriptions = load_yaml_config_file(
            os.path.join(os.path.dirname(__file__), 'services.yaml'))[DOMAIN]
        hass.services.register(DOMAIN, SERVICE_SEND_COMMAND, _tx,
                               descriptions[SERVICE_SEND_COMMAND],
                               SERVICE_SEND_COMMAND_SCHEMA)
        hass.services.register(DOMAIN, SERVICE_VOLUME, _volume,
                               descriptions[SERVICE_VOLUME],
                               SERVICE_VOLUME_SCHEMA)
        hass.services.register(DOMAIN, SERVICE_UPDATE_DEVICES, _update,
                               descriptions[SERVICE_UPDATE_DEVICES],
                               SERVICE_UPDATE_DEVICES_SCHEMA)
        hass.services.register(DOMAIN, SERVICE_POWER_ON, _power_on)
        hass.services.register(DOMAIN, SERVICE_STANDBY, _standby)
        hass.services.register(DOMAIN, SERVICE_SELECT_DEVICE, _select_device)

        hdmi_network.set_new_device_callback(_new_device)
        hdmi_network.start()

    hass.bus.listen_once(EVENT_HOMEASSISTANT_START, _start_cec)
    hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, _shutdown)
    return True


class CecDevice(Entity):
    """Representation of a HDMI CEC device entity."""

    def __init__(self, hass: HomeAssistant, device, logical):
        """Initialize the device."""
        self._device = device
        self.hass = hass
        self._icon = None
        self._state = STATE_UNKNOWN
        self._logical_address = logical
        self.entity_id = "%s.%d" % (DOMAIN, self._logical_address)
        device.set_update_callback(self._update)

    def update(self):
        """Update device status."""
        self._update()

    def _update(self, device=None):
        """Update device status."""
        if device:
            from pycec.const import STATUS_PLAY, STATUS_STOP, STATUS_STILL, \
                POWER_OFF, POWER_ON
            if device.power_status == POWER_OFF:
                self._state = STATE_OFF
            elif device.status == STATUS_PLAY:
                self._state = STATE_PLAYING
            elif device.status == STATUS_STOP:
                self._state = STATE_IDLE
            elif device.status == STATUS_STILL:
                self._state = STATE_PAUSED
            elif device.power_status == POWER_ON:
                self._state = STATE_ON
            else:
                _LOGGER.warning("Unknown state: %d", device.power_status)
        self.schedule_update_ha_state()

    @property
    def name(self):
        """Return the name of the device."""
        return (
            "%s %s" % (self.vendor_name, self._device.osd_name)
            if (self._device.osd_name is not None and
                self.vendor_name is not None and self.vendor_name != 'Unknown')
            else "%s %d" % (self._device.type_name, self._logical_address)
            if self._device.osd_name is None
            else "%s %d (%s)" % (self._device.type_name, self._logical_address,
                                 self._device.osd_name))

    @property
    def vendor_id(self):
        """ID of device's vendor."""
        return self._device.vendor_id

    @property
    def vendor_name(self):
        """Name of device's vendor."""
        return self._device.vendor

    @property
    def physical_address(self):
        """Physical address of device in HDMI network."""
        return str(self._device.physical_address)

    @property
    def type(self):
        """String representation of device's type."""
        return self._device.type_name

    @property
    def type_id(self):
        """Type ID of device."""
        return self._device.type

    @property
    def icon(self):
        """Icon for device by its type."""
        return (self._icon if self._icon is not None else
                ICONS_BY_TYPE.get(self._device.type)
                if self._device.type in ICONS_BY_TYPE else ICON_UNKNOWN)

    @property
    def device_state_attributes(self):
        """Return the state attributes."""
        state_attr = {}
        if self.vendor_id is not None:
            state_attr[ATTR_VENDOR_ID] = self.vendor_id
            state_attr[ATTR_VENDOR_NAME] = self.vendor_name
        if self.type_id is not None:
            state_attr[ATTR_TYPE_ID] = self.type_id
            state_attr[ATTR_TYPE] = self.type
        if self.physical_address is not None:
            state_attr[ATTR_PHYSICAL_ADDRESS] = self.physical_address
        return state_attr