evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add remote control platform to BraviaTV (#50845)

Browse source
This commit is contained in:
Artem Draft 2021-06-17 13:33:44 +03:00 • committed by GitHub
parent 08b0ef7a5e
commit db61a773fd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 384 additions and 197 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -122,6 +122,7 @@ omit =
homeassistant/components/braviatv/__init__.py
homeassistant/components/braviatv/const.py
homeassistant/components/braviatv/media_player.py
homeassistant/components/braviatv/remote.py
homeassistant/components/broadlink/__init__.py
homeassistant/components/broadlink/const.py
homeassistant/components/broadlink/remote.py

2
CODEOWNERS
View file

@ -72,7 +72,7 @@ homeassistant/components/bmp280/* @belidzs
homeassistant/components/bmw_connected_drive/* @gerard33 @rikroe
homeassistant/components/bond/* @prystupa
homeassistant/components/bosch_shc/* @tschamm
homeassistant/components/braviatv/* @bieniu
homeassistant/components/braviatv/* @bieniu @Drafteed
homeassistant/components/broadlink/* @danielhiversen @felipediel
homeassistant/components/brother/* @bieniu
homeassistant/components/brunt/* @eavanvalkenburg

246
homeassistant/components/braviatv/__init__.py
View file

@ -1,24 +1,47 @@
"""The Bravia TV component."""
import asyncio
from datetime import timedelta
import logging
from bravia_tv import BraviaRC
from bravia_tv.braviarc import NoIPControl
from homeassistant.const import CONF_HOST, CONF_MAC
from homeassistant.components.media_player import DOMAIN as MEDIA_PLAYER_DOMAIN
from homeassistant.components.remote import DOMAIN as REMOTE_DOMAIN
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import BRAVIARC, DOMAIN, UNDO_UPDATE_LISTENER
from .const import (
BRAVIA_COORDINATOR,
CLIENTID_PREFIX,
CONF_IGNORED_SOURCES,
DOMAIN,
NICKNAME,
UNDO_UPDATE_LISTENER,
)
PLATFORMS = ["media_player"]
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [MEDIA_PLAYER_DOMAIN, REMOTE_DOMAIN]
SCAN_INTERVAL = timedelta(seconds=10)
async def async_setup_entry(hass, config_entry):
"""Set up a config entry."""
host = config_entry.data[CONF_HOST]
mac = config_entry.data[CONF_MAC]
pin = config_entry.data[CONF_PIN]
ignored_sources = config_entry.options.get(CONF_IGNORED_SOURCES, [])
coordinator = BraviaTVCoordinator(hass, host, mac, pin, ignored_sources)
undo_listener = config_entry.add_update_listener(update_listener)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][config_entry.entry_id] = {
BRAVIARC: BraviaRC(host, mac),
BRAVIA_COORDINATOR: coordinator,
UNDO_UPDATE_LISTENER: undo_listener,
}
@ -44,3 +67,218 @@ async def async_unload_entry(hass, config_entry):
async def update_listener(hass, config_entry):
"""Handle options update."""
await hass.config_entries.async_reload(config_entry.entry_id)
class BraviaTVCoordinator(DataUpdateCoordinator[None]):
"""Representation of a Bravia TV Coordinator.
An instance is used per device to share the same power state between
several platforms.
"""
def __init__(self, hass, host, mac, pin, ignored_sources):
"""Initialize Bravia TV Client."""
self.braviarc = BraviaRC(host, mac)
self.pin = pin
self.ignored_sources = ignored_sources
self.muted = False
self.program_name = None
self.channel_name = None
self.channel_number = None
self.source = None
self.source_list = []
self.original_content_list = []
self.content_mapping = {}
self.duration = None
self.content_uri = None
self.start_date_time = None
self.program_media_type = None
self.audio_output = None
self.min_volume = None
self.max_volume = None
self.volume = None
self.is_on = False
# Assume that the TV is in Play mode
self.playing = True
self.state_lock = asyncio.Lock()
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
request_refresh_debouncer=Debouncer(
hass, _LOGGER, cooldown=1.0, immediate=False
),
)
def _send_command(self, command, repeats=1):
"""Send a command to the TV."""
for _ in range(repeats):
for cmd in command:
self.braviarc.send_command(cmd)
def _get_source(self):
"""Return the name of the source."""
for key, value in self.content_mapping.items():
if value == self.content_uri:
return key
def _refresh_volume(self):
"""Refresh volume information."""
volume_info = self.braviarc.get_volume_info(self.audio_output)
if volume_info is not None:
self.audio_output = volume_info.get("target")
self.volume = volume_info.get("volume")
self.min_volume = volume_info.get("minVolume")
self.max_volume = volume_info.get("maxVolume")
self.muted = volume_info.get("mute")
return True
return False
def _refresh_channels(self):
"""Refresh source and channels list."""
if not self.source_list:
self.content_mapping = self.braviarc.load_source_list()
self.source_list = []
if not self.content_mapping:
return False
for key in self.content_mapping:
if key not in self.ignored_sources:
self.source_list.append(key)
return True
def _refresh_playing_info(self):
"""Refresh playing information."""
playing_info = self.braviarc.get_playing_info()
self.program_name = playing_info.get("programTitle")
self.channel_name = playing_info.get("title")
self.program_media_type = playing_info.get("programMediaType")
self.channel_number = playing_info.get("dispNum")
self.content_uri = playing_info.get("uri")
self.source = self._get_source()
self.duration = playing_info.get("durationSec")
self.start_date_time = playing_info.get("startDateTime")
if not playing_info:
self.channel_name = "App"
def _update_tv_data(self):
"""Connect and update TV info."""
power_status = self.braviarc.get_power_status()
if power_status != "off":
connected = self.braviarc.is_connected()
if not connected:
try:
connected = self.braviarc.connect(
self.pin, CLIENTID_PREFIX, NICKNAME
)
except NoIPControl:
_LOGGER.error("IP Control is disabled in the TV settings")
if not connected:
power_status = "off"
if power_status == "active":
self.is_on = True
if self._refresh_volume() and self._refresh_channels():
self._refresh_playing_info()
return
self.is_on = False
async def _async_update_data(self):
"""Fetch the latest data."""
if self.state_lock.locked():
return
await self.hass.async_add_executor_job(self._update_tv_data)
async def async_turn_on(self):
"""Turn the device on."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.turn_on)
await self.async_request_refresh()
async def async_turn_off(self):
"""Turn off device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.turn_off)
await self.async_request_refresh()
async def async_set_volume_level(self, volume):
"""Set volume level, range 0..1."""
async with self.state_lock:
await self.hass.async_add_executor_job(
self.braviarc.set_volume_level, volume, self.audio_output
)
await self.async_request_refresh()
async def async_volume_up(self):
"""Send volume up command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(
self.braviarc.volume_up, self.audio_output
)
await self.async_request_refresh()
async def async_volume_down(self):
"""Send volume down command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(
self.braviarc.volume_down, self.audio_output
)
await self.async_request_refresh()
async def async_volume_mute(self, mute):
"""Send mute command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.mute_volume, mute)
await self.async_request_refresh()
async def async_media_play(self):
"""Send play command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_play)
self.playing = True
await self.async_request_refresh()
async def async_media_pause(self):
"""Send pause command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_pause)
self.playing = False
await self.async_request_refresh()
async def async_media_stop(self):
"""Send stop command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_stop)
self.playing = False
await self.async_request_refresh()
async def async_media_next_track(self):
"""Send next track command."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_next_track)
await self.async_request_refresh()
async def async_media_previous_track(self):
"""Send previous track command."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_previous_track)
await self.async_request_refresh()
async def async_select_source(self, source):
"""Set the input source."""
if source in self.content_mapping:
uri = self.content_mapping[source]
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.play_content, uri)
await self.async_request_refresh()
async def async_send_command(self, command, repeats):
"""Send command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self._send_command, command, repeats)
await self.async_request_refresh()

7
homeassistant/components/braviatv/config_flow.py
View file

@ -16,7 +16,7 @@ from .const import (
ATTR_CID,
ATTR_MAC,
ATTR_MODEL,
BRAVIARC,
BRAVIA_COORDINATOR,
CLIENTID_PREFIX,
CONF_IGNORED_SOURCES,
DOMAIN,
@ -160,7 +160,10 @@ class BraviaTVOptionsFlowHandler(config_entries.OptionsFlow):
async def async_step_init(self, user_input=None):
"""Manage the options."""
self.braviarc = self.hass.data[DOMAIN][self.config_entry.entry_id][BRAVIARC]
coordinator = self.hass.data[DOMAIN][self.config_entry.entry_id][
BRAVIA_COORDINATOR
]
self.braviarc = coordinator.braviarc
connected = await self.hass.async_add_executor_job(self.braviarc.is_connected)
if not connected:
await self.hass.async_add_executor_job(

2
homeassistant/components/braviatv/const.py
View file

@ -6,7 +6,7 @@ ATTR_MODEL = "model"
CONF_IGNORED_SOURCES = "ignored_sources"
BRAVIARC = "braviarc"
BRAVIA_COORDINATOR = "bravia_coordinator"
BRAVIA_CONFIG_FILE = "bravia.conf"
CLIENTID_PREFIX = "HomeAssistant"
DEFAULT_NAME = f"{ATTR_MANUFACTURER} Bravia TV"

2
homeassistant/components/braviatv/manifest.json
View file

@ -3,7 +3,7 @@
"name": "Sony Bravia TV",
"documentation": "https://www.home-assistant.io/integrations/braviatv",
"requirements": ["bravia-tv==1.0.11"],
"codeowners": ["@bieniu"],
"codeowners": ["@bieniu", "@Drafteed"],
"config_flow": true,
"iot_class": "local_polling"
}

252
homeassistant/components/braviatv/media_player.py
View file

@ -1,8 +1,6 @@
"""Support for interface with a Bravia TV."""
import asyncio
import logging
from bravia_tv.braviarc import NoIPControl
import voluptuous as vol
from homeassistant.components.media_player import (
@ -24,19 +22,24 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_STEP,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PIN, STATE_OFF, STATE_ON
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PIN,
STATE_OFF,
STATE_PAUSED,
STATE_PLAYING,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util.json import load_json
from .const import (
ATTR_MANUFACTURER,
BRAVIA_CONFIG_FILE,
BRAVIARC,
CLIENTID_PREFIX,
CONF_IGNORED_SOURCES,
BRAVIA_COORDINATOR,
DEFAULT_NAME,
DOMAIN,
NICKNAME,
)
_LOGGER = logging.getLogger(__name__)
@ -94,9 +97,9 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Add BraviaTV entities from a config_entry."""
ignored_sources = []
pin = config_entry.data[CONF_PIN]
"""Set up Bravia TV Media Player from a config_entry."""
coordinator = hass.data[DOMAIN][config_entry.entry_id][BRAVIA_COORDINATOR]
unique_id = config_entry.unique_id
device_info = {
"identifiers": {(DOMAIN, unique_id)},
@ -105,135 +108,25 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
"model": config_entry.title,
}
braviarc = hass.data[DOMAIN][config_entry.entry_id][BRAVIARC]
ignored_sources = config_entry.options.get(CONF_IGNORED_SOURCES, [])
async_add_entities(
[
BraviaTVDevice(
braviarc, DEFAULT_NAME, pin, unique_id, device_info, ignored_sources
)
]
[BraviaTVMediaPlayer(coordinator, DEFAULT_NAME, unique_id, device_info)]
)
class BraviaTVDevice(MediaPlayerEntity):
"""Representation of a Bravia TV."""
class BraviaTVMediaPlayer(CoordinatorEntity, MediaPlayerEntity):
"""Representation of a Bravia TV Media Player."""
_attr_device_class = DEVICE_CLASS_TV
_attr_supported_features = SUPPORT_BRAVIA
def __init__(self, client, name, pin, unique_id, device_info, ignored_sources):
"""Initialize the Bravia TV device."""
def __init__(self, coordinator, name, unique_id, device_info):
"""Initialize the entity."""
self._pin = pin
self._braviarc = client
self._name = name
self._state = STATE_OFF
self._muted = False
self._program_name = None
self._channel_name = None
self._channel_number = None
self._source = None
self._source_list = []
self._original_content_list = []
self._content_mapping = {}
self._duration = None
self._content_uri = None
self._playing = False
self._start_date_time = None
self._program_media_type = None
self._audio_output = None
self._min_volume = None
self._max_volume = None
self._volume = None
self._unique_id = unique_id
self._device_info = device_info
self._ignored_sources = ignored_sources
self._state_lock = asyncio.Lock()
async def async_update(self):
"""Update TV info."""
if self._state_lock.locked():
return
power_status = await self.hass.async_add_executor_job(
self._braviarc.get_power_status
)
if power_status != "off":
connected = await self.hass.async_add_executor_job(
self._braviarc.is_connected
)
if not connected:
try:
connected = await self.hass.async_add_executor_job(
self._braviarc.connect, self._pin, CLIENTID_PREFIX, NICKNAME
)
except NoIPControl:
_LOGGER.error("IP Control is disabled in the TV settings")
if not connected:
power_status = "off"
if power_status == "active":
self._state = STATE_ON
if (
await self._async_refresh_volume()
and await self._async_refresh_channels()
):
await self._async_refresh_playing_info()
return
self._state = STATE_OFF
def _get_source(self):
"""Return the name of the source."""
for key, value in self._content_mapping.items():
if value == self._content_uri:
return key
async def _async_refresh_volume(self):
"""Refresh volume information."""
volume_info = await self.hass.async_add_executor_job(
self._braviarc.get_volume_info, self._audio_output
)
if volume_info is not None:
self._audio_output = volume_info.get("target")
self._volume = volume_info.get("volume")
self._min_volume = volume_info.get("minVolume")
self._max_volume = volume_info.get("maxVolume")
self._muted = volume_info.get("mute")
return True
return False
async def _async_refresh_channels(self):
"""Refresh source and channels list."""
if not self._source_list:
self._content_mapping = await self.hass.async_add_executor_job(
self._braviarc.load_source_list
)
self._source_list = []
if not self._content_mapping:
return False
for key in self._content_mapping:
if key not in self._ignored_sources:
self._source_list.append(key)
return True
async def _async_refresh_playing_info(self):
"""Refresh Playing information."""
playing_info = await self.hass.async_add_executor_job(
self._braviarc.get_playing_info
)
self._program_name = playing_info.get("programTitle")
self._channel_name = playing_info.get("title")
self._program_media_type = playing_info.get("programMediaType")
self._channel_number = playing_info.get("dispNum")
self._content_uri = playing_info.get("uri")
self._source = self._get_source()
self._duration = playing_info.get("durationSec")
self._start_date_time = playing_info.get("startDateTime")
if not playing_info:
self._channel_name = "App"
super().__init__(coordinator)
@property
def name(self):
@ -253,113 +146,96 @@ class BraviaTVDevice(MediaPlayerEntity):
@property
def state(self):
"""Return the state of the device."""
return self._state
if self.coordinator.is_on:
return STATE_PLAYING if self.coordinator.playing else STATE_PAUSED
return STATE_OFF
@property
def source(self):
"""Return the current input source."""
return self._source
return self.coordinator.source
@property
def source_list(self):
"""List of available input sources."""
return self._source_list
return self.coordinator.source_list
@property
def volume_level(self):
"""Volume level of the media player (0..1)."""
if self._volume is not None:
return self._volume / 100
if self.coordinator.volume is not None:
return self.coordinator.volume / 100
return None
@property
def is_volume_muted(self):
"""Boolean if volume is currently muted."""
return self._muted
@property
def supported_features(self):
"""Flag media player features that are supported."""
return SUPPORT_BRAVIA
return self.coordinator.muted
@property
def media_title(self):
"""Title of current playing media."""
return_value = None
if self._channel_name is not None:
return_value = self._channel_name
if self._program_name is not None:
return_value = f"{return_value}: {self._program_name}"
if self.coordinator.channel_name is not None:
return_value = self.coordinator.channel_name
if self.coordinator.program_name is not None:
return_value = f"{return_value}: {self.coordinator.program_name}"
return return_value
@property
def media_content_id(self):
"""Content ID of current playing media."""
return self._channel_name
return self.coordinator.channel_name
@property
def media_duration(self):
"""Duration of current playing media in seconds."""
return self._duration
def set_volume_level(self, volume):
"""Set volume level, range 0..1."""
self._braviarc.set_volume_level(volume, self._audio_output)
return self.coordinator.duration
async def async_turn_on(self):
"""Turn the media player on."""
async with self._state_lock:
await self.hass.async_add_executor_job(self._braviarc.turn_on)
"""Turn the device on."""
await self.coordinator.async_turn_on()
async def async_turn_off(self):
"""Turn off media player."""
async with self._state_lock:
await self.hass.async_add_executor_job(self._braviarc.turn_off)
"""Turn the device off."""
await self.coordinator.async_turn_off()
def volume_up(self):
"""Volume up the media player."""
self._braviarc.volume_up(self._audio_output)
async def async_set_volume_level(self, volume):
"""Set volume level, range 0..1."""
await self.coordinator.async_set_volume_level(volume)
def volume_down(self):
"""Volume down media player."""
self._braviarc.volume_down(self._audio_output)
async def async_volume_up(self):
"""Send volume up command."""
await self.coordinator.async_volume_up()
def mute_volume(self, mute):
async def async_volume_down(self):
"""Send volume down command."""
await self.coordinator.async_volume_down()
async def async_mute_volume(self, mute):
"""Send mute command."""
self._braviarc.mute_volume(mute)
await self.coordinator.async_volume_mute(mute)
def select_source(self, source):
async def async_select_source(self, source):
"""Set the input source."""
if source in self._content_mapping:
uri = self._content_mapping[source]
self._braviarc.play_content(uri)
await self.coordinator.async_select_source(source)
def media_play_pause(self):
"""Simulate play pause media player."""
if self._playing:
self.media_pause()
else:
self.media_play()
def media_play(self):
async def async_media_play(self):
"""Send play command."""
self._playing = True
self._braviarc.media_play()
await self.coordinator.async_media_play()
def media_pause(self):
"""Send media pause command to media player."""
self._playing = False
self._braviarc.media_pause()
async def async_media_pause(self):
"""Send pause command."""
await self.coordinator.async_media_pause()
def media_stop(self):
async def async_media_stop(self):
"""Send media stop command to media player."""
self._playing = False
self._braviarc.media_stop()
await self.coordinator.async_media_stop()
def media_next_track(self):
async def async_media_next_track(self):
"""Send next track command."""
self._braviarc.media_next_track()
await self.coordinator.async_media_next_track()
def media_previous_track(self):
"""Send the previous track command."""
self._braviarc.media_previous_track()
async def async_media_previous_track(self):
"""Send previous track command."""
await self.coordinator.async_media_previous_track()

69
homeassistant/components/braviatv/remote.py Normal file
View file

@ -0,0 +1,69 @@
"""Remote control support for Bravia TV."""
from homeassistant.components.remote import ATTR_NUM_REPEATS, RemoteEntity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTR_MANUFACTURER, BRAVIA_COORDINATOR, DEFAULT_NAME, DOMAIN
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up Bravia TV Remote from a config entry."""
coordinator = hass.data[DOMAIN][config_entry.entry_id][BRAVIA_COORDINATOR]
unique_id = config_entry.unique_id
device_info = {
"identifiers": {(DOMAIN, unique_id)},
"name": DEFAULT_NAME,
"manufacturer": ATTR_MANUFACTURER,
"model": config_entry.title,
}
async_add_entities(
[BraviaTVRemote(coordinator, DEFAULT_NAME, unique_id, device_info)]
)
class BraviaTVRemote(CoordinatorEntity, RemoteEntity):
"""Representation of a Bravia TV Remote."""
def __init__(self, coordinator, name, unique_id, device_info):
"""Initialize the entity."""
self._name = name
self._unique_id = unique_id
self._device_info = device_info
super().__init__(coordinator)
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._unique_id
@property
def device_info(self):
"""Return device specific attributes."""
return self._device_info
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def is_on(self):
"""Return true if device is on."""
return self.coordinator.is_on
async def async_turn_on(self, **kwargs):
"""Turn the device on."""
await self.coordinator.async_turn_on()
async def async_turn_off(self, **kwargs):
"""Turn the device off."""
await self.coordinator.async_turn_off()
async def async_send_command(self, command, **kwargs):
"""Send a command to device."""
repeats = kwargs[ATTR_NUM_REPEATS]
await self.coordinator.async_send_command(command, repeats)